/* * 1.Mapper 類的四個(gè)泛型中，前兩個(gè)指定 mapper 輸入數(shù)據(jù)的類型,后兩個(gè)指定 mapper 輸出數(shù)據(jù)的類型 *   KEYIN 是輸入的 key 的類型，VALUEIN 是輸入的 value 的類型 *   KEYOUT 是輸出的 key 的類型，VALUEOUT 是輸出的 value 的類型 * 2.map 和 reduce 的數(shù)據(jù)的輸入輸出都是以 key-value 對(duì)的形式封裝的 * 3.默認(rèn)情況下，框架傳遞給我們的 mapper 的輸入數(shù)據(jù)中，key 是要處理的文本中一行的起始偏移量，為 Long 類型， * 這一行的內(nèi)容為 value，為 String 類型的  * 4.后兩個(gè)泛型的賦值需要我們結(jié)合實(shí)際情況 * 5.為了在網(wǎng)絡(luò)中傳輸時(shí)序列化更高效，Hadoop 把 Java 中的 Long 封裝為 LongWritable, 把 String 封裝為 Text */public class WCMapper extends Mapper<LongWritable, Text, Text, LongWritable> {        //重寫 Mapper 中的 map 方法，MapReduce 框架每讀一行數(shù)據(jù)就調(diào)用一次此方法    @Override    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {        //書寫具體的業(yè)務(wù)邏輯，業(yè)務(wù)要處理的數(shù)據(jù)已經(jīng)被框架傳遞進(jìn)來(lái)，就是方法的參數(shù)中的 key 和 value        //key 是這一行數(shù)據(jù)的起始偏移量，value 是這一行的文本內(nèi)容                //1.將 Text 類型的一行的內(nèi)容轉(zhuǎn)為 String 類型        String line = value.toString();                //2.使用 StringUtils 以空格切分字符串，返回 String[]        String[] words = StringUtils.split(line, " ");                //3.循環(huán)遍歷 String[],調(diào)用 context 的 writer()方法，輸出為 key-value 對(duì)的形式        //key：?jiǎn)卧~   value：1        for(String word : words) {            context.write(new Text(word), new LongWritable(1));        }            }}

/* * 1.Reducer 類的四個(gè)泛型中，前兩個(gè)輸入要與 Mapper 的輸出相對(duì)應(yīng)。輸出需要聯(lián)系具體情況自定義 */public class WCReducer extends Reducer<Text, LongWritable, Text, LongWritable> {        //框架在 map 處理完之后，將所有的 kv 對(duì)緩存起來(lái)，進(jìn)行分組，然后傳遞一個(gè)分組(<key,{values}>,例如：<"hello",{1,1,1,1}>),    //調(diào)用此方法    @Override    protected void reduce(Text key, Iterable<LongWritable> values, Context context)throws IOException, InterruptedException {                //1.定義一個(gè)計(jì)數(shù)器        long count = 0;                //2.遍歷 values的 list，進(jìn)行累加求和        for(LongWritable value : values) {            //使用 LongWritable 的 get() 方法，可以將 一個(gè) LongWritable 類型轉(zhuǎn)為 Long 類型            count += value.get();        }                //3.輸出這一個(gè)單詞的統(tǒng)計(jì)結(jié)果        context.write(key, new LongWritable(count));    }}

/** * 此類用來(lái)描述一個(gè)特定的作業(yè) * 例：1.該作業(yè)使用哪個(gè)類作為邏輯處理中的 map，哪個(gè)作為 reduce *       2.指定該作業(yè)要處理的數(shù)據(jù)所在的路徑 *       3.指定該作業(yè)輸出的結(jié)果放到哪個(gè)路徑 */public class WCRunner {        public static void main(String[] args) throws Exception {                //1.獲取 Job 對(duì)象:使用 Job 靜態(tài)的 getInstance() 方法，傳入 Configuration 對(duì)象        Configuration conf = new Configuration();        Job wcJob = Job.getInstance(conf);                //2.設(shè)置整個(gè) Job 所用的類的 jar 包：使用 Job 的 setJarByClass(),一般傳入  當(dāng)前類.class        wcJob.setJarByClass(WCRunner.class);                //3.設(shè)置本 Job 使用的 mapper 和 reducer 的類        wcJob.setMapperClass(WCMapper.class);        wcJob.setReducerClass(WCReducer.class);                        //4.指定 reducer 輸出數(shù)據(jù)的 kv 類型  注：若 mapper 和 reducer 的輸出數(shù)據(jù)的 kv 類型一致，可以用如下兩行代碼設(shè)置        wcJob.setOutputKeyClass(Text.class);        wcJob.setOutputValueClass(LongWritable.class);                //5.指定 mapper 輸出數(shù)據(jù)的 kv 類型        wcJob.setMapOutputKeyClass(Text.class);        wcJob.setMapOutputValueClass(LongWritable.class);                //6.指定原始的輸入數(shù)據(jù)存放路徑:使用 FileInputFormat 的 setInputPaths() 方法        FileInputFormat.setInputPaths(wcJob, new Path("/wc/srcdata/"));                //7.指定處理結(jié)果的存放路徑：使用 FileOutputFormat 的 setOutputFormat() 方法        FileOutputFormat.setOutputPath(wcJob, new Path("/wc/output/"));                //8.將 Job 提交給集群運(yùn)行，參數(shù)為 true 表示顯示運(yùn)行狀態(tài)        wcJob.waitForCompletion(true);            }}