MapReduce编程之Partitioner

Partitioner存在的意义

image.png

Partitioner 处理示意图

我们可以指定 Reduce 处理的数据 按照一定的 规则或者方式 进行分组

image.png

验证 Partitioner的测试数据格式如下：

我们想要实现的就是相同类型的手机放入到一个Reduce里面执行输出 。

image.png

修改代码如下：

加入了 Partitioner
以及修改了 Map 以及Driver 部分代码
运行 发现 输出四个文件 并且每个文件有不同类型的类型的计算

package com.zyh.hadoop;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

/**
 * 使用MapReduce开发WordCount用用程序
 * 相比 WordCountApp3 增加了  Partition
 */
public class WordCountApp4 {
    /**
     * 读取输入文件
     * <p>
     * Mapper
     * LongWritable 偏移量
     */
    public static class MyMapper extends Mapper<LongWritable, Text, Text, LongWritable> {
        LongWritable one = new LongWritable(1);
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            // 接收到的 每一行数据
            String line = value.toString();
            // 按照制定分隔符分开
            String[] words = line.split(" ");
            context.write(new Text(words[0]), new LongWritable(Long.parseLong(words[1])));
        }
    }

    /**
     * 归并操作
     */
    public static class MyReducer extends Reducer<Text, LongWritable, Text, LongWritable> {
        @Override
        protected void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {
            long sum = 0;
            for (LongWritable value : values) {
                sum += value.get();
            }
            context.write(key, new LongWritable(sum));
        }
    }

    public static class MyPartitioner extends Partitioner<Text, LongWritable> {

        @Override
        public int getPartition(Text text, LongWritable longWritable, int numPartitions) {
            if (text.toString().equals("xiaomi")) {
                return 0;
            }
            if (text.toString().equals("huawei")) {
                return 1;
            }
            if (text.toString().equals("iphone7")) {
                return 2;
            }
            return 3;
        }
    }
    /**
     * 封装了所有的mapreduce的 作业cd
     * @param args
     */
    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        //创建configuration
        Configuration configuration = new Configuration();

        //校验是否已存在 输出的目录文件  存在即删除
        Path path = new Path(args[1]);
        FileSystem fileSystem = FileSystem.get(configuration);
        if (fileSystem.exists(path)) {
            fileSystem.delete(path, true);
            System.out.println("output file  delete success ");
        }
        
        Job job = Job.getInstance(configuration, "wordCount");
        //设置 job的处理类
        job.setJarByClass(WordCountApp4.class);
        //设置作业处理的输入路径
        FileInputFormat.setInputPaths(job, new Path(args[0]));
        //设置map相关的参数
        job.setMapperClass(MyMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(LongWritable.class);


        //设置 reduce 相关参数
        job.setReducerClass(MyReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(LongWritable.class);


        //设置  job的 partition
        job.setPartitionerClass(MyPartitioner.class);
        // 设置 4个 reducer 每个分区 一个
        job.setNumReduceTasks(4);

        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}