Hadoop ChainMap

单一MapReduce对一些非常简单的问题提供了很好的支持。但是如果处理过程变得更加复杂，这种复杂性应该体现为更多地MapReduce工作，而不是更加复杂的map函数和reduce函数。

 

在hadoop 中一个Job中可以按顺序运行多个mapper对数据进行前期的处理，再进行reduce，经reduce后的结果可经个经多个按顺序执行的mapper进行后期的处理，这样的Job是不会保存中间结果的，并大大减少了I/O操作。

 

例如：在一个Job中，按顺序执行 Map1->Map2->Reduce->Map3->Map4 ，在这种链式结构中，要将Map2与Reduce看成这个MapReduce的核心部分，partitioning与shuffling(奇迹发生的地方)在此处才会被应用到。所以Map1作为前期处理，而Map3与Map4作为后期处理。

 

代码示例:

 

Configuration conf = getConf();
JobConf job = new JobConf(conf);

job.setJobName(“ChainJob”);
job.setInputFormat(TextInputFormat.class);
job.setOutputFormat(TextOutputFormat.class);

FileInputFormat.setInputPaths(job, in);
FileOutputFormat.setOutputPath(job, out);

JobConf map1Conf = new JobConf(false);

//将map1加入到Job中
ChainMapper.addMapp(job,
                    Map1.class,
                    LongWritable.class, 
                    Text.class,
                    Text.class,
                    Text.class,
                    true,
                    map1Conf);

//将map2加入到Job中
JobConf map2Conf = new JobConf(false);
ChainMapper.addMapper(job,
                      Map.class,
                      Text.class,
                      Text.class, 
                      LongWritable.class,
                      Text.class,
                      true,
                      map2Conf);

//将Reduce加入到Job中
JobConf reduceConf = new JobConf(false);
ChainReducer.setReducer(job,
                        Reduce.class,
                        LongWritable.class,
                        Text.class,
                        Text.class,
                        Text.class,
                        true,
                        reduceConf);

//将map3加入到Job中
JobConf map3Conf = new JobConf(false);
ChainReducer.addMapper(job,
                       Map3.class,
                       Text.class,
                       Text.class,
                       LongWritable.class, 
                       Text.class,
                       true,
                       map3Conf);

//将map4加入到Job中
JobConf map4Conf = new JobConf(false);
ChainReducer.addMapper(job,
                       Map4.class,
                       LongWritable.class,
                       Text.class, 
                       LongWritable.class,
                       Text.class,
                       true,
                       map4Conf);


JobClient.runJob(job);

 

 

 

addMapper方法的签名(setReducer方法与此类似)

 

public static <K1,V1,K2,V2> void

addMapper(JobConf job,

                    Class<? extends Mapper<K1,V1,K2,V2>> klass,

                    Class<? extends K1> inputKeyClass,

                    Class<? extends V1> inputValueClass,

                    Class<? extends K2> outputKeyClass,

                    Class<? extends V2> outputValueClass,

                    boolean byValue,

                    JobConf mapperConf)

 

关于byValue参数：

 

在标准的Mapper模型中，键/值对的输出在序列化之后写入磁盘，等待被shuffer到一个可能完全不同的节点上。形式上认为这个过程采用的是值传递。发送的是键/值对副本。在目前的情况下，我们可以将一个Mapper与另一个相链接，在相同的JVM线程中一起执行。因此，键/值对的发送有可能采用引用传递，初始Mapper的输出放在内存中，后续的Mapper直接引用相同的内存位置。当Map1调用context.write(K k, V v)时，对象k和v直接传递给Map2的map方法。mapper之间可能有大量的数据需要传递，避免复制这些数据可以让性能得以提高。但是这么做违背了Hadoop中MapReduce Model的一个微妙约定：对Context.write(K k, V v)的调用一定不会改变k和v的内容。

如果我们可以确信Map1的map()方法在调用Context.write(k, v)之后不会再使用k和v的内容，或者Map2并不改变k和v在其上的输入值，可以通过设定byValue为false来获得一定的性能提升。如果我们对Mapper的内部代码不太了解，最好依旧采用值传递，确保Mapper会按预期的方式工作。

 

评论

3 楼 brandNewUser 2014-09-18  

楼主你好，问个问题，为什么我写的如下的：
JobConf phase1JobConf = new JobConf(false);
        ChainMapper.addMapper(jobConf, Phase1Mapper.class, Text.class, History.class, Phase1KeyDecorator.class,
                BytesWritable.class, true, phase1JobConf);

        JobConf phase2ReducerConf = new JobConf(false);
        ChainReducer.setReducer(jobConf, Phase1Reducer.class, Phase1KeyDecorator.class, BytesWritable.class,
                Text.class, Text.class, true, phase2ReducerConf);

        JobConf phase3ChainJobConf = new JobConf(false);
        ChainMapper.addMapper(jobConf, Phase3ChainMapper.class, Text.class, Text.class, Phase2KeyDecorator.class,
                BytesWritable.class, true, phase3ChainJobConf);

会抛出错误：


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)

Hadoop2.2.0版本，提示在检查的时候Key/Value类型错误，难道是中间的Reducer类型没起作用？

2 楼 Genie13 2012-05-29  

有个问题，如何咋0.20.2这个版本实现ChainMApper
好奇怪啊，这个版本Job  Mapper Reducer 是新的api  可是ChainMapper ChainReducer 确是旧的api

0.22.0 这个版本是新的ChainMapper ChainReducer
可是1.0的这个又变成旧的ChainMapper ChainReducer
好奇怪啊

1 楼 chenwq 2012-05-27  

http://www.distream.org/?p=379