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);

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏沃趣科技

MySQL排序内部原理探秘

一、我们要解决什么问题 二、排序,排序,排序 三、索引优化排序 四、排序模式 4.1实际trace结果 4.2排序模式概览 4.2.1回表排序模式 4.2.2不...

4866
来自专栏耕耘实录

Linux三大剑客之awk

版权声明:本文为耕耘实录原创文章,各大自媒体平台同步更新。欢迎转载,转载请注明出处,谢谢

1624
来自专栏coolblog.xyz技术专栏

自己动手实现一个简单的JSON解析器

JSON(JavaScript Object Notation) 是一种轻量级的数据交换格式。相对于另一种数据交换格式 XML,JSON 有着诸多优点。比如易读...

67619
来自专栏python学习路

三、模型(一)

当我们的程序涉及到数据库相关操作时,我们一般都会这么做: 创建数据库,设计表结构和字段 使用 MySQLdb 来连接数据库,并编写数据访问层代码 业务逻辑层去调...

4079
来自专栏PHP在线

mysql学习笔记

InnoDB引擎与MyISAM引擎 mysql是关系型数据库。其中的存储引擎可以show engines来查看。我的版本是5.6.26的,查看版本用selec...

2855
来自专栏Android知识点总结

自己写一个svg转化为安卓xml的工具类

3072
来自专栏Golang语言社区

go语言的sql包原理与用法分析

go的sql包是在pkg/database中,里面的两个包sql和sql/driver可以一起看。建议看这个两个包之前可以先看看sql文件夹下的doc.txt。...

3014
来自专栏Java3y

阅读SSM项目之scm

导入项目 项目是由eclipse来编写的,我使用的开发环境是Idea,那么就需要将eclipse项目导入进去Idea中。要想项目能够启动起来。是这样干的: 导入...

34711
来自专栏zhisheng

《从0到1学习Flink》—— 如何自定义 Data Source ?

在 《从0到1学习Flink》—— Data Source 介绍 文章中,我给大家介绍了 Flink Data Source 以及简短的介绍了一下自定义 Dat...

1394
来自专栏Golang语言社区

go语言的sql包原理与用法分析

本文实例讲述了go语言的sql包原理与用法。分享给大家供大家参考,具体如下: go的sql包是在pkg/database中,里面的两个包sql和sql/driv...

4496

扫码关注云+社区

领取腾讯云代金券