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

 

1.迭代式mapreduce

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

      在MapReduce的迭代思想，类似for循环，前一个 MapReduce的输出结果，作为下一个 MapReduce的输入，任务完成后中间结果都可以删除。

代码示例：

 

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

 

 

 

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

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

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

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

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

代码示例:

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

 

 

2.依赖关系组合式MapReduce

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

伪代码示例:

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

 

3.链式MapReduce

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

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

 

 

 

参考:MapReduce，组合式，迭代式，链式

 

 

评论

1 楼 chenwq 2012-05-27  

http://blog.csdn.net/RiverM?viewmode=contents