MapReduce-WorldCount编程思路

这里要与Spark中的map和reduce算子做区分。Spark的map()和reduce()与Hadoop MapReduce中的同名函数没有直接对应关系。

张三 1
张三 1
李四 1
王五 1
赵六 1
李四 1
李四 1

张三 [1,1]
李四 [1,1,1]
王五 [1]
赵六 [1]

张三 2
李四 3
王五 1
赵六 1

/*
	继承Mapper类，只要输入的是文字，泛型中有固定的数据类型：
	k-LongWritable		== L
	v-Text				== String
    不固定的是输出的 k-v，在该示例中：
	人名-k-Text
	1-v-IntWritable	 	== Integer
*/
static class WordCountMapper extends Mapper<LongWritable, Text, Text, IntWritable> {//输入输入、输出输出

    @Override
    protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
      //Mapper类根据每个逻辑切片中的数据行数，调用对应次数的map方法，源码中是while循环
        String s = value.toString();
        String[] arr = s.split(" ");	// 将人名按照空格分隔并写入数组arr
        for (String name : arr) {
            // 遍历迭代器，利用context工具类，将人名写入k，1写入v
            context.write(new Text(name),new IntWritable(1));
        }
    }
}

static class WordCountReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
//map的输出是什么，reduce的输入就是什么，泛型可以直接复制
    @Override
    protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
        int sum = 0;
        //对合并后的文件进行遍历，需要一个累加器
        for (IntWritable value : values) {
            int i = value.get();	//每一个value表示其中的一个1，循环通过get()方法转为int类型
            sum += i;
        }
        context.write(key, new IntWritable(sum));
    }
}
// 输出
// 张三	12
// 李四	6
// 王五	6

// 1. 初始化配置
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", "hdfs://hadoop10:8020");
        //2. 创建job
        Job job = Job.getInstance(conf);
        job.setJarByClass(CountCleanJob.class);

        //3. 设置输入格式化工具和输出格式化
        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);

        //4. 设置输入路径和输出路径
    	// TextInputFormat读取hdfs中的文件，根据文件大小和个数对其split逻辑切片
		// MapReduce根据切片多少，启动对应的MapTask任务
        TextInputFormat.addInputPath(job, new Path("/wordcount.txt"));
    	// 将输出文件发送到hdfs
        TextOutputFormat.setOutputPath(job, new Path("/Count_Out"));

        //5. 设置mapper和reducer
        job.setMapperClass(CountCleanMapper.class);
        job.setReducerClass(CountCleanReducer.class);

        // 6. 设置mapper的kv类型和reducer的kv类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);

        // 7. 启动job
        boolean b = job.waitForCompletion(true);
        System.out.println(b);//true
        ```