问题描述

我们的数据分析平台是单一的Map/Reduce过程，由于半年来不断地增加需求，导致了问题已经不是那么地简单，特别是在Reduce阶段，一些大对象会常驻内存。因此越来越顶不住压力了，当前内存问题已经是最大的问题，每个Map占用5G，每个Reduce占用9G！直接导致当数据分析平台运行时，集群处于资源匮乏状态。

因此，在不改变业务数据计算的条件下，将单一的Map/Reduce过程分解成2个阶段。这个时候，需求就相对来说比较复杂，将第一阶段的Reduce结果输出至HDFS，作为第二阶段的输入。

其基本过程图很简单如下所示：

我们可以使用启动两个Job，在第一个阶段Job完成之后，再进行第二阶段Job的执行。但是更好的方式是使用Hadoop中提供的JobControl工具，这个工具可以加入多个等待执行的子Job，并定义其依赖关系，决定执行的先后顺序。

JobControl jobControl = new JobControl(GROUP_NAME);

        JobConf phase1JobConf = Phase1Main.getJobConf(getConf(), jsonConfigFilePhase1, reduceCountOne);

        Job phase1Job = new Job(phase1JobConf);

        jobControl.addJob(phase1Job);

        JobConf phase2JobConf = Phase2Main.getJobConf(getConf(), jsonConfigFilePhase2, reduceCountTwo);

        Job phase2Job = new Job(phase2JobConf);

        jobControl.addJob(phase2Job);

        phase2Job.addDependingJob(phase1Job);

        jobControl.run();

        return jobControl.getFailedJobs() == null || jobControl.getFailedJobs().isEmpty() ? 0 : 1;

正如代码所示，可以使用job的addDependingJob(JobConf)方法来定义其依赖关系。

但是这种方式有一个非常大的缺点，如果中间数据结果过大，将其放置在HDFS上是非常浪费磁盘资源，同时也带来后续过多的I/O操作，包括第一阶段的写磁盘和第二阶段的读磁盘（而且本身中间结果数据也没有什么太大用途）。

经过查阅，在Hadoop中，一个Job可以按顺序执行多个mapper对数据进行前期的处理，再进行Reduce，Reduce执行完成后，还可以继续执行多个Mapper，形成一个处理链结构，这样的Job是不会存储中间结果的，大大减少了磁盘I/O操作。

但这种方式也对map/reduce程序有个要求，就是只能存在一个Partition规则，因为整个链条中只会存在一次Reduce操作。前文介绍的那两个阶段的Partition规则如果不一致，是不能改造成这种方式的。

这种方式的大致流程图如下：

由于我们的分析程序中，第二步就需要根据一定的规则进行聚集，因此第二步就需要进行Reduce，将原来第四步的Reduce阶段强行改造成Map阶段。注意，Map阶段之间互相传递数据时，其数量是固定的，而且不会进行聚集（Reduce）操作，还是需要按照流的方式进行处理，因此最好要先排序。单个Map的结果只会传递给特定的单个下个步骤的Map端。

在ChainMain类中会执行这种方式，需要借助于ChainMapper和ChainReducer两个Hadoop中提供的类：

String finalJobName = TongCommonConstants.JOB_NAME + jobNameSuffix;

        jobConf.setJobName(finalJobName);

        jobConf.setInputFormat(RawLogInputFormat.class);

        jobConf.setPartitionerClass(Phase1Partitioner.class);

        jobConf.setNumReduceTasks(reduceCountTwo);

        jobConf.set(TongCommonConstants.DIC_INFO, jsonConfigFile);

        DicInfoManager.getInstance().readDicManager(jobConf, jsonConfigFile);

        String yesterdayOutDir = DicInfoManager.getInstance().getDicManager().getPrevious_day_output_path();

        JobConf phase1JobConf = getJobConf(jsonConfigFile, yesterdayOutDir);

        ChainMapper.addMapper(jobConf, Phase1Mapper.class, Text.class, History.class, Phase1KeyDecorator.class,

                BytesWritable.class, true, phase1JobConf);

        JobConf phase2ReducerConf = getJobConf(jsonConfigFile, yesterdayOutDir);

        ChainReducer.setReducer(jobConf, Phase1Reducer.class, Phase1KeyDecorator.class, BytesWritable.class,

                Text.class, Text.class, true, phase2ReducerConf);

        JobConf phase3ChainJobConf = getJobConf(jsonConfigFile, yesterdayOutDir);

        ChainReducer.addMapper(jobConf, Phase3ChainMapper.class, Text.class, Text.class, Phase2KeyDecorator.class,

                BytesWritable.class, true, phase3ChainJobConf);

        JobConf phase4ChainJobConf = getJobConf(jsonConfigFile, yesterdayOutDir);

        ChainReducer.addMapper(jobConf, Phase4ChainMapper.class, Phase2KeyDecorator.class, BytesWritable.class,

                Text.class, Text.class, true, phase4ChainJobConf);

        RunningJob runningJob = JobClient.runJob(jobConf);

        runningJob.waitForCompletion();

        return runningJob.isSuccessful() ? 0 : 1;

经过这种方式的改造后，对原有程序的影响最小，因为不需要定义中间结果存储地址，当然也不需要定义第二阶段的配置文件。

新手比较容易犯的一个错误是，Reducer后面的map步骤要使用ChainReducer.addMapper方法而不是ChainMapper.addMapper方法，否则会抱下面的异常，我就在这个上面栽了跟头，查了很久。

Exception in thread "main" java.lang.IllegalArgumentException: The specified Mapper input key class does not match the previous Mapper's output key class.

at org.apache.hadoop.mapreduce.lib.chain.Chain.validateKeyValueTypes(Chain.java:695)

at org.apache.hadoop.mapred.lib.Chain.addMapper(Chain.java:104)

Hadoop工作流中的JobControl

很多情况下，用户编写的作业比较复杂，相互之间存在依赖关系，这种可以用有向图表示的依赖关系称之为“工作流”。

JobControl是由两个类组成：Job和JobControl，Job的状态转移图如下：


 

作业在刚开始的时候处于Waiting状态，如果没有依赖作业或者所有依赖作业都已经完成的情况下，进入Ready状态；一旦进入Ready状态，则作业可被提交到Hadoop集群上运行，并进入Running状态，根据作业的运行情况，可能进入Success或Failed状态。需要注意的是，如果一个作业的依赖作业失败，则该作业也会失败，后续的所有作业也都会失败。

JobControl封装了一系列MapReduce作业及其对应的依赖关系，它将处于不同状态的作业放入不同的哈希表，按照Job的状态转移图转移作业，直到所有作业运行完成。在实现的时候，JobControl包含一个线程用于周期性地监控和更新各个作业的运行状态，调度依赖作业运行完成的作业，提交Ready状态的作业等。

ChainMapper/ChainReduce

ChainMapper/ChainReducer主要是为了解决线性链式Mapper而提出的，在Map或Reduce阶段存在多个Mapper，像多个Linux管道一样，前一个Mapper的输出结果直接重定向到下一个Mapper的输入，形成一个流水线，最后的Mapper或Reducer才会将结果写到HDFS上。对于任意一个MapReduce作业，Map和Reduce阶段可以由无限个Mapper，但只能有一个Reducer。

Hadoop MapReduce有一个约定，函数OutputCollector.collect(key, value)执行期间不能改变key和value的值，这是因为某个map/reduce调用该方法之后，可能后续继续再次使用key和value的值，如果被改变，可能会造成潜在的错误。

ChainMapper/Reducer实现的关键技术点就是修改Mapper和Reducer的输出流，将本来要写入文件的输出结果重定向到另外一个Mapper中。尽管链式作业在Map和Reduce阶段添加了多个Mapper，但仍然只是一个MapReduce作业，因而只能有一个与之对应的JobConf对象。

ChainMapper中实现的map函数大概如下：

public void map(Object key, Object value, OutputCollector output, Reporter reporter) throws IOException{

     Mapper mapper = chain.getFirstMap();

     if(mapper != null){

          mapper.map(key, value, chain.getMapperCollector(0, output, reporter), reporter);

     }

}

chain.getMapperCollector返回一个OutputCollector实现，即ChainOutputCollector，collector方法大概如下：

public void collect(K key, V value) throws IOException{

     if(nextMapperIndex < mappers.size()){

          //调用下一个Mapper，直到没有mapper

          nextMapper.map(key, value, new ChainOutputCollector(nextMapperIndex, nextKeySerialization, nextValueSerialization, output, reporter));

     } else {

          //如果是最后一个Mapper，直接调用真正的Collector

          output.collect(key, value);

     }

}

在使用ChainMapper/ChainReducer时需要注意一个问题：就是其中参数byValue的选择，究竟是该传值还是传递引用。因为在Hadoop编程中需要处理的数据量比较大，经常使用复用同一个对象的情况，普通的Mapper/Reducer程序由于不会执行链式处理，在其他的JVM中来重建Map输出的对象，而Chain API中需要管道一样的操作来进行下一步处理，Mapper.map()函数调用完outputCollector.collect(key, value)之后，可能再次使用key和value的值，才导致这个问题的发生。

个人总觉得虽然重用引用的方式虽然可以节省一定的内存，但是不重用引用也仅仅会对Minor GC造成一定的压力，如果严格控制生成的new对象Key，Value的生命周期的话。

正是为了防止OutputCollector直接对key/value进行修改，ChainMapper允许用户指定key/value的传递方式，如果编写的程序确定key/value执行期间不会被重用以修改（如果是不可变对象最好），则可以选择按照引用来进行传递，否则按值传递。需要注意的是，引用传递可以避免对象的深层拷贝，提高处理效率，但需要编程时做出key/value不能修改的保证。