MapReduce中实现自定义分区Partitioner

心已赠人 2021-12-13 04:28 305阅读 0赞

有时候在利用mapreduce进行任务计算时，需要按照不同的规则，将不同的结果输出到不同的文件中，以便将计算结果分类。比如有这样一组数据，我们需要根据第一列的编号进行划分，让相同编号的第二列内容输出到同一个文件中，不同编号的内容输出到不同的文件中。

0,hello world
    1,hello ketty
    2,hello tom
    0,hello lyf
    0,good morning
    2,test
    3,33333

**方案一：MultipleOutputs类**

利用MapReduce提供的MultipleOutputs类（API 2.0）可以很容易的实现。MultipleOutputs类提供了三种不同的write函数用于将结果输出到不同的文件中，可根据自己的情况灵活使用。

public <K, V> void write(String namedOutput, K key, V value)
    public <K, V> void write(String namedOutput, K key, V value, String baseOutputPath)
    public void write(KEYOUT key, VALUEOUT value, String baseOutputPath)
    
    参数：
        key:               输出的key
        Value:             输出的value
        namedOutput:       在JOB中定义的输出名，可自定义不同的输出格式
        baseOutputPath:    自定义输出文件目录，可为多级目录，如果最后一级不存在，则默认为输出文件的前缀名

无奈集群某些原因，任务迁移至别的集群之后MultiOutputs类无法使用，只好另寻出路了，于是就有了下面的方案。

**方案二：自定义Partitioner**

利用MapReduce提供的Partitioner类，我们可以自定义我们自己的Partitioner。

首先我们需要了解MapReduce的工作原理，我们知道，简单来说Mapper阶段主要是负责分发数据到各个Reducer，然后Reducer负责聚合各个Mapper分发过来的数据。一般情况下，这个具体如何分发的我们并不太关心，这是因为框架本身有一个默认的设置，会帮我们自动完成。而这个默认的自定如何分发数据的类就是Partitioner。我们来看一下默认的HashPartitioner。

package org.apache.hadoop.mapreduce.lib.partition;
    
    import org.apache.hadoop.classification.InterfaceAudience.Public;
    import org.apache.hadoop.classification.InterfaceStability.Stable;
    import org.apache.hadoop.mapreduce.Partitioner;
    
    @Public
    @Stable
    public class HashPartitioner<K, V> extends Partitioner<K, V> {
        public HashPartitioner() {
        }
    
        public int getPartition(K key, V value, int numReduceTasks) {
            return (key.hashCode() & 2147483647) % numReduceTasks;
        }
    }

从源码我们可以看出，HashPartitioner包含一个getPartition函数，这个函数就是定义数据分发到哪个reduce的。参数key和value就是我们的map的输出，numReduceTasks就是我们定义的reduce个数，默认为1。当reduce个数设置为1时，任何数除以1的余数都是0，所有的数据都会发送给编号为0的reduce。看到这我想大家就知道如何自定义的控制结果输出了，只要我们能够让不同的key和value对应不同的reduce编号即可，即让getPartition函数返回0，1，2，3…就可以（注意：前提是设置的reduce个数必须大于getPartition返回的最大值）。

**自定义Partitioner**

假设我们让能被3整除的输出到编号为0的reduce里，余数为1的输出到编号为1到reduce里，余数为2的输出到编号为2的reduce里，代码如下：

import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Partitioner;
    
    public class MyDummyPartitions extends Partitioner<IntWritable, Text> {
    
        @Override
        public int getPartition(IntWritable intWritable, Text text, int i) {
            return intWritable.get() % 3;
        }
    }

最终的结果将会如下：

part-r-00000文件

good morning
    hello lyf
    hello world
    33333

part-r-00001文件

hello ketty

part-r-00002文件

test
    hello tom

**完整代码如下：**

DummyPartitionsMapper类

import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.LongWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Mapper;
    
    import java.io.IOException;
    
    public class DummyPartitionsMapper extends Mapper<LongWritable, Text, IntWritable, Text> {
    
        private IntWritable outputKey;
        private Text outputVal;
    
        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            outputKey = new IntWritable();
            outputVal = new Text();
        }
    
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
    
            System.out.println(value);
    
            String[] val= value.toString().split(",");
            int index = Integer.parseInt(val[0]);
    
            outputKey.set(index);
            outputVal.set(new Text(val[1]));
    
            context.write(outputKey, outputVal);
        }
    }

DummyPartitionsReducer类

import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.NullWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Reducer;
    
    import java.io.IOException;
    
    public class DummyPartitionsReducer extends Reducer<IntWritable, Text, NullWritable, Text> {
    
        @Override
        protected void reduce(IntWritable key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
    
            for (Text val: values) {
                context.write(NullWritable.get(), val);
            }
        }
    }

自定义MyDummyPartitions类

DummyPartitionsJob类

import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.conf.Configured;
    import org.apache.hadoop.fs.Path;
    import org.apache.hadoop.io.IntWritable;
    import org.apache.hadoop.io.NullWritable;
    import org.apache.hadoop.io.Text;
    import org.apache.hadoop.mapreduce.Job;
    import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
    import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
    import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
    import org.apache.hadoop.mapreduce.lib.output.LazyOutputFormat;
    import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
    import org.apache.hadoop.util.Tool;
    import org.apache.hadoop.util.ToolRunner;
    
    import java.util.HashMap;
    import java.util.Map;
    
    public class DummyPartitionsJob extends Configured implements Tool {
    
        @Override
        public int run(String[] strings) throws Exception {
    
            Configuration conf = getConf();
    
            String inputPath = "data_input";
            String outputPath = "data_output";
    
            if (HDFSUtil.exists(conf, outputPath)){
                HDFSUtil.rm(conf, outputPath, true);
            }
    
            Job job = Job.getInstance(conf);
            job.setJarByClass(DummyPartitionsJob.class);
    
            job.setMapperClass(DummyPartitionsMapper.class);
            job.setMapOutputKeyClass(IntWritable.class);
            job.setMapOutputValueClass(Text.class);
    
            job.setReducerClass(DummyPartitionsReducer.class);
            job.setOutputKeyClass(NullWritable.class);
            job.setOutputValueClass(Text.class);
            job.setNumReduceTasks(3);                               // 设置reduce个数
    
            job.setPartitionerClass(MyDummyPartitions.class);       // 指定自定义的Partitioner
    
            job.setInputFormatClass(TextInputFormat.class);
            job.setOutputFormatClass(LazyOutputFormat.class);
    
            LazyOutputFormat.setOutputFormatClass(job, TextOutputFormat.class);
    
            FileInputFormat.addInputPath(job, new Path(inputPath));
            FileOutputFormat.setOutputPath(job, new Path(outputPath));
    
            boolean ret = job.waitForCompletion(true);
            return ret? 0: 1;
        }
    
        public static void main(String[] args) throws Exception {
            System.exit(ToolRunner.run(new DummyPartitionsJob(), args));
        }
    }