MapReduce：N keys，N files（一）

【背景】

MapReduce中，不管是map阶段还是reduce阶段，二者的输入和输出都是key，value类型的值。现在有个需求是根据map阶段返回值key的个数，生成相应个数的文件。也就说一个key写到一个文件中，每个文件只能包含一个key。

这种需求存在两种情况：
1、keys是固定的一组数，因此reduce的个数是确定的
2、keys是不固定的，但reduce的个数是确定的

两种情况对应两个解决方案，下面对这两种情况一一解释。

【keys是固定的一组数】

keys是固定的一组数，说的是keys只出现在几个固定的数值中。比如说省份，手机号码前三位等等。解决这类问题只需要更改Partitioner和设置Reducer的个数即可。
Partitioner负责将map输出的key重新划分，分配给reduce。其getPartition方法就是获取key分配给哪个reducer，其范围值是[0, len(numOfReducer)-1]。MapReduce默认是采用HashPartitioner进行分区。

举例：假设有六种电话号码，编写partitioner
Partitioner：

    package com.hadoop.mapreduce;
    
    import java.util.HashMap;
    import java.util.Map;
    
    import org.apache.hadoop.io.NullWritable;
    import org.apache.hadoop.mapreduce.lib.partition.HashPartitioner;
    
    import entity.UserEntity;
    
    /*
     * Partitioner用于划分键值空间（key space）。
     * Partitioner组件可以让Map对Key进行分区，从而可以根据不同的key来分发到不同的reduce中去处理。
     * 分区的数量与一个作业的reduce任务的数量是一样的。
     * 它控制将中间过程的key（也就是这条记录）应该发送给m个reduce任务中的哪一个来进行reduce操作。
     * HashPartitioner是默认的 Partitioner。 
     */
    
    /**
     * 继承抽象类Partitioner，实现自定义的getPartition()方法
     * 通过job.setPartitionerClass()来设置自定义的Partitioner；
     */
    public class ProviderPartitioner extends HashPartitioner<UserEntity, NullWritable> {
        
        // 声明providerMap，并且在static静态块中初始化
        private static Map<String, Integer> providerMap = new HashMap<String, Integer>();
        static {
            providerMap.put("130", 0);
            providerMap.put("133", 1);
            providerMap.put("134", 2);
            providerMap.put("135", 3);
            providerMap.put("136", 4);
            providerMap.put("137", 5);
        }
    
        /**
         * 实现自定义的getPartition()方法,自定义分区规则
         */
        @Override
        public int getPartition(UserEntity key, NullWritable value, int numPartitions) {
            String prefix = key.getMobile().substring(0, 3);
            return providerMap.get(prefix);
        }
    }

在job中设置分区类和配置reducer个数：

// 设置定义分区的处理类
job.setPartitionerClass(ProviderPartitioner.class);
job.setNumReduceTasks(6);

详细内容可参考：https://blog.csdn.net/yuan_xw/article/details/50867819

【keys一直变化】

这种情况下，不能确定具体的keys，只知道keys的个数是一个确定的值。比如说一流互联网公司有BAT，TMD等，但我们只需取出排名前500即可。
这种情况下由于key是一直变的，也就说排名前500的互联网公司是一直变的，改写partitioner无法满足需求。由于同一个key肯定是全部分配到一个reducer中的（一个reducer接收的不止一个key），我们可以通过修改输出类型（outputFormat），将不同的key分配到指定文件即可。

以修改OrcOutputFormat为例：

package is.split;

import org.apache.commons.lang.RandomStringUtils;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.orc.OrcFile;
import org.apache.orc.Writer;
import org.apache.orc.mapred.OrcStruct;
import org.apache.orc.mapreduce.OrcMapreduceRecordWriter;
import org.apache.orc.mapreduce.OrcOutputFormat;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

public class OrcReNameFileOutputFormat extends OrcOutputFormat {

    @Override
    public RecordWriter<Text, OrcStruct> getRecordWriter(TaskAttemptContext taskAttemptContext) throws IOException {
        return new OrcReNameMapreduceRecordWriter(taskAttemptContext);
    }


    private class OrcReNameMapreduceRecordWriter extends RecordWriter<Text, OrcStruct>{
        //保存各个key对应文件的writer
        private Map<String, OrcMapreduceRecordWriter> map;
        //private OrcMapreduceRecordWriter realWrite ;
        private TaskAttemptContext taskAttemptContext;

        public OrcReNameMapreduceRecordWriter(TaskAttemptContext taskAttemptContext){
            this.taskAttemptContext = taskAttemptContext;
            this.map = new HashMap<String, OrcMapreduceRecordWriter>();
        }

        //该函数接收的key是map阶段输出的key
        public void write(Text key, OrcStruct value) throws IOException, InterruptedException {
          //真正向文件中写数据的Writer，还是OrcMapreduceRecordWriter
            OrcMapreduceRecordWriter realWrite = map.get(key.toString());
            if (realWrite == null){
                //String outputFileName = taskAttemptContext.getConfiguration().get("ouputfile.prefix.col.name", "uiappid");
            
                String split_key =  key.toString();
                //输出路径文件夹，以该key为文件夹
                String outputDirPath = taskAttemptContext.getConfiguration().get(FileOutputFormat.OUTDIR ) + "/" + split_key ;
                //输出路径文件名，文件名是根据当前时间戳和六位随机数生成的
                Path filename = new Path(new Path(outputDirPath), ISTool.getCurTime() + "_" + RandomStringUtils.randomAlphanumeric(6) );
                Writer writer = OrcFile.createWriter(filename, org.apache.orc.mapred.OrcOutputFormat.buildOptions(taskAttemptContext.getConfiguration()));
                realWrite = new OrcMapreduceRecordWriter<OrcStruct>(writer);
                map.put(key.toString(), realWrite);
            }
            realWrite.write(NullWritable.get(), value);
        }

        public void close(TaskAttemptContext context) throws IOException, InterruptedException {
            for (Map.Entry<String, OrcMapreduceRecordWriter> entry : map.entrySet()) {
                if (entry.getValue() != null){
                    entry.getValue().close(context);
                }

            }
        }
    }
}

在job中设置reducer个数和reduce OutputFormat。

job.setNumReduceTasks(500);
job.setOutputFormatClass(OrcReNameFileOutputFormat.class);