MapReduce，组合式，迭代式，链式

小咪咪 2022-03-12 03:24 205阅读 0赞

前面介绍一些怎样用户类制定自己的类，来达到减少中间数据：[http://www.cnblogs.com/liqizhou/archive/2012/05/14/2499498.html][http_www.cnblogs.com_liqizhou_archive_2012_05_14_2499498.html]

## 1.迭代式mapreduce ##

一些复杂的任务难以用一次mapreduce处理完成，需要多次mapreduce才能完成任务，例如Pagrank，Kmeans算法都需要多次的迭代，关于mapreduce迭代在mahout中运用较多。有兴趣的可以参考一下mahout的源码。

在map/reduce迭代过程中，思想还是比较简单，就像类似for循环一样，前一个mapreduce的输出结果，作为下一个mapreduce的输入，任务完成后中间结果都可以删除。如代码所以：

![复制代码][copycode.gif]

Configuration conf1 = new Configuration();
    Job job1 = new Job(conf1,"job1");
    .....
    FileInputFormat.addInputPath(job1,InputPaht1);
    FileOutputFromat.setOoutputPath(job1,Outpath1);
    job1.waitForCompletion(true);
    //sub Mapreduce
    Configuration conf2 = new Configuration();
    Job job2 = new Job(conf1,"job1");
    .....
    FileInputFormat.addInputPath(job2,Outpath1);
    FileOutputFromat.setOoutputPath(job2,Outpath2);
    job2.waitForCompletion(true);
    //sub Mapreduce
    Configuration conf3 = new Configuration();
    Job job3 = new Job(conf1,"job1");
    .....
    FileInputFormat.addInputPath(job3,Outpath2);
    FileOutputFromat.setOoutputPath(job3,Outpath3);
    job3.waitForCompletion(true);
    .....

![复制代码][copycode.gif]

下面列举一个mahout怎样运用mapreduce迭代的，下面的代码快就是mahout中kmeans的算法的代码，在main函数中用一个while循环来做mapreduce的迭代，其中：runIteration()是一次mapreduce的过程。

但个人感觉现在的mapreduce迭代设计不太满意的地方。

1. 每次迭代，如果所有Job（task）重复创建，代价将非常高。

2.每次迭代，数据都写入本地和读取本地，I/O和网络传输的代价比较大。

好像Twister和Haloop的模型能过比较好的解决这些问题，但他们抽象度不够高，支持的计算有限。

期待着下个版本hadoop更好的支持迭代算法。

![复制代码][copycode.gif]

//main function
    while (!converged && iteration <= maxIterations) {
          log.info("K-Means Iteration {}", iteration);
          // point the output to a new directory per iteration
          Path clustersOut = new Path(output, AbstractCluster.CLUSTERS_DIR + iteration);
          converged = runIteration(conf, input, clustersIn, clustersOut, measure.getClass().getName(), delta);
          // now point the input to the old output directory
          clustersIn = clustersOut;
          iteration++;
    }
    
      private static boolean runIteration(Configuration conf,
                                          Path input,
                                          Path clustersIn,
                                          Path clustersOut,
                                          String measureClass,
                                          String convergenceDelta)
        throws IOException, InterruptedException, ClassNotFoundException {
    
        conf.set(KMeansConfigKeys.CLUSTER_PATH_KEY, clustersIn.toString());
        conf.set(KMeansConfigKeys.DISTANCE_MEASURE_KEY, measureClass);
        conf.set(KMeansConfigKeys.CLUSTER_CONVERGENCE_KEY, convergenceDelta);
    
        Job job = new Job(conf, "KMeans Driver running runIteration over clustersIn: " + clustersIn);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(ClusterObservations.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Cluster.class);
    
        job.setInputFormatClass(SequenceFileInputFormat.class);
        job.setOutputFormatClass(SequenceFileOutputFormat.class);
        job.setMapperClass(KMeansMapper.class);
        job.setCombinerClass(KMeansCombiner.class);
        job.setReducerClass(KMeansReducer.class);
    
        FileInputFormat.addInputPath(job, input);
        FileOutputFormat.setOutputPath(job, clustersOut);
    
        job.setJarByClass(KMeansDriver.class);
        HadoopUtil.delete(conf, clustersOut);
        if (!job.waitForCompletion(true)) {
          throw new InterruptedException("K-Means Iteration failed processing " + clustersIn);
        }
        FileSystem fs = FileSystem.get(clustersOut.toUri(), conf);
    
        return isConverged(clustersOut, conf, fs);
      }

![复制代码][copycode.gif]

## 2.依赖关系组合式MapReduce ##

我们可以设想一下MapReduce有3个子任务job1,job2,job3构成，其中job1和job2相互独立，job3要在job1和job2完成之后才执行。这种关系就叫复杂数据依赖关系的组合时mapreduce。hadoop为这种组合关系提供了一种执行和控制机制，hadoop通过job和jobControl类提供具体的编程方法。Job除了维护子任务的配置信息，还维护子任务的依赖关系，而jobControl控制整个作业流程，把所有的子任务作业加入到JobControl中，执行JobControl的run（）方法即可运行程序。

下面给出伪代码：

![复制代码][copycode.gif]

Configuration job1conf = new Configuration();
    Job job1 = new Job(job1conf,"Job1");
    .........//job1 其他设置
    Configuration job2conf = new Configuration();
    Job job2 = new Job(job2conf,"Job2");
    .........//job2 其他设置
    Configuration job3conf = new Configuration();
    Job job3 = new Job(job3conf,"Job3");
    .........//job3 其他设置
    job3.addDepending(job1);//设置job3和job1的依赖关系
    job3.addDepending(job2);
    JobControl JC = new JobControl("123");
    JC.addJob(job1);//把三个job加入到jobcontorl中
    JC.addJob(job2);
    JC.addJob(job3);
    JC.run();

![复制代码][copycode.gif]

## 3.链式MapReduce ##

首先看一下例子，来说明为什么要有链式MapReduce，假设在统计单词是，会出现这样的词，make，made，making等，他们都属于一个词，在单词累加的时候，都归于一个词。解决的方法为用一个单独的Mapreduce任务可以实现，单增加了多个Mapreduce作业，将增加整个作业处理的周期，还增加了I/O操作，因而处理效率不高。

一个较好的办法就是在核心的MapReduce之外，增加一个辅助的Map过程，然后将这个辅助的Map过程和核心的Mapreudce过程合并为一个链式的Mapreduce，从而完成整个作业。hadoop提供了专门的链式ChainMapper和ChainReducer来处理链式任务，ChainMapper允许一个Map任务中添加多个Map的子任务，ChainReducer可以在Reducer执行之后，在加入多个Map的子任务。其调用形式如下：

![复制代码][copycode.gif]

ChainMapper.addMapper(...);
        ChainReducer.addMapper(...);
        //addMapper()调用的方法形式如下：
        public static void addMapper(JOb job,
                Class<? extends Mapper> mclass,
                Class<?> inputKeyClass,
                Class<?> inputValueClass,
                Class<?> outputKeyClass,
                Class<?> outputValueClass,
                Configuration conf
        ){
        }

![复制代码][copycode.gif]

其中，ChainReducer专门提供了一个setRreducer()方法来设置整个作业唯一的Reducer。

note：这些Mapper和Reducer之间传递的键和值都必须保持一致。

下面举个例子：用ChainMapper把Map1加如并执行，然后用ChainReducer把Reduce和Map2加入到Reduce过程中。代码如下：Map1.class 要实现map方法

![复制代码][copycode.gif]

public void function throws IOException {
            Configuration conf = new Configuration();
            Job job = new Job(conf);
            job.setJobName("ChianJOb");
            // 在ChainMapper里面添加Map1
            Configuration map1conf = new Configuration(false);
            ChainMapper.addMapper(job, Map1.class, LongWritable.class, Text.class,
                    Text.class, Text.class, true, map1conf);
            // 在ChainReduce中加入Reducer，Map2；
            Configuration reduceConf = new Configuration(false);
            ChainReducer.setReducer(job, Reduce.class, LongWritable.class,
                    Text.class, Text.class, Text.class, true, map1conf);
            Configuration map2Conf = new Configuration();
            ChainReducer.addMapper(job, Map2.class, LongWritable.class, Text.class,
                    Text.class, Text.class, true, map1conf);
            job.waitForCompletion(true);
        }

--------------------

[http_www.cnblogs.com_liqizhou_archive_2012_05_14_2499498.html]: http://www.cnblogs.com/liqizhou/archive/2012/05/14/2499498.html
[copycode.gif]: /images/20220312/2c410682e8744d2495ee1f15863a8151.png