環境說明:

• jdk1.8
• hadoop-2.7.7
• windows上配置好的帶有hadoop環境的eclipse

1.自定義Mapper

/*
 * LongWritable對應輸入的key型別，預設是行的偏移量LongWritable
 * Text,對應上輸入的value型別，預設行資料Text
 * Text:對應輸出的key型別，不能使用預設值，需要根據需求更改
 * Text:對應輸出的value型別，根據需求修改
 * @author lesie
 * 要求輸出的格式(key,1)
 * 單詞計數輸出的key型別為Text
 * 輸出的value型別為IntWritable
 
 
*/
public class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable>{
    
    /*
     * KEYIN
     * VALUEIN
     * context--環境物件，輸出結果
     * @see org.apach.hadoop.mapreduce.Mapper#map(KEYIN,VALUEIN,...)
     */
    public void map(LongWritable ikey,Text ivalue,Context context) throws
 
 IOException, InterruptedException
    {
        //獲取一行資料
        String line=ivalue.toString();
        
        //按空格切片
        String []arrs=line.split(" ");
        for(String arr:arrs)
        {
            context.write(new Text(arr),new IntWritable(1));
        }
        
    }
}

2.自定義Reducer

/*
 

 * reducer的數輸入key用公式mapper輸出的key型別
 * valuein:reducer的輸入value應該是mapper輸出的value型別
 * keyout:根據業務而定
 * valueout:根據業務而定
 * @author lesie 
 * 工作機制:
 * 1.將key相同的value進行合併，形成一個Iterable，交給程式
 * eg:(hello,<1,1,1,1,1,1>)
 * 2.reduce方法執行的次數取決於mapper輸出的key，有多個不同的key執行多少次
 * 3.預設的排序，對key進行排序，先按照數字進行排再按照字典順序
 */

public class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
    
    public void reduce(Text _key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        // process values
        //定義計數變數
        int sum=0;
        //進行累加操作
        for (IntWritable val : values) {
            //通過get方法取出其中的值
            sum+=val.get();
        }
        //輸出資料,最終結果,key是單詞Text,value是單詞出現的總次數
        context.write(_key, new IntWritable(sum));
    }

}

3.主程式

public class WordCountDriver {

    public static void main(String[] args) throws Exception {
        // TODO Auto-generated method stub
        //獲取當前配置
        Configuration conf=new Configuration();
        
        //獲取一個表示當前Mapreduce作業的Job物件，向ahdoop申請一個job任務執行邏輯
        Job job=Job.getInstance();
        
        //指定程式入口
        job.setJarByClass(WordCountDriver.class);
        
        //設定需要執行的Mapper類
        job.setMapperClass(WordCountMapper.class);
        
        //設定Reducer類
        job.setReducerClass(WordCountReducer.class);
        
        //設定Mapper的輸出型別
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);
        
        //設定Reducer的輸出結果型別
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        
        //設定輸入路徑
        FileInputFormat.setInputPaths(job, new Path("hdfs://192.168.140.128:9000/wc/words.txt"));
        
        //設定輸出路徑
        FileOutputFormat.setOutputPath(job, new Path("hdfs://192.168.140.128:9000/wc/result6"));
        
        //任務的提交
        job.waitForCompletion(true);
    }

}