本帖最后由 xuanxufeng 于 2015-8-27 21:59 编辑
问题导读
1.什么链式mapreduce？
2.链式mapreduce解决了什么问题？
3.如何实现链式mapreduce?
相关文章：让你真正明白什么是MapReduce组合式，迭代式，链式
http://www.aboutyun.com/thread-7435-1-1.html
链接MapReduce作业
1. 顺序链接MapReduce作业
顺序链接MapReduce作业就是将多个MapReduce作业作为生成的一个自动化执行序列，将上一个MapReduce作业的输出作为下一个MapReduce作业的输入。MapReduce作业的链接就类似于Unix 的管道：
mapreduce – 1 | mapreduce – 2 | mapreduce – 3 | ···
顺序链接MapReduce作业的执行过程，就是driver为MapReduce作业创建一个带有配置参数的JobConf对象，并将该对象传递给JobClient.runJob()来启动这个作业。而当JobClient.runJob()运行到作业结尾处时会被阻止，MapReduce作业的链接就会在这个时候调用另外一个MapReduce作业的driver。每个作业的driver都必须创建一个新的JobConf对象，并将其输入路径设置为前一个作业的输出路径。
2. 具有复杂依赖的MapReduce链接
如果两个或者多个MapReduce作业之间存在执行的先后顺序关系的话，那么这个时候就不能用顺序链接。Hadoop如何处理这种MapReduce作业之间的关系的呢，Hadoop通过Job和JobControl类来管理这种（非线性）作业之间的依赖。
Job对象是MapReduce作业的表现形式。Job对象的实例化通过传递一个JobConf对象到作业的构造函数中来实现，也就是说通过public Job(JobConf jobConf){}来实现实例化一个MapReduce作业。Job通过Job1.addDependingJob(Job2)来维护作业间的依赖关系这也意味着Job1在Job2完成之前是不会启动的。
JobControl对象负责管理并监视作业Job的执行。添加作业：addJob()方法。而当所有作业和依赖关系添加完成之后，调用JobControl的run()方法，就会生成一个专门提交作业并监视作业执行的线程，这样就实现了JobControl对Job的管理和监视。
3. 预处理和后处理阶段的链接
Hadoop关于预处理和后处理作业的链接提供了两种解决方案。
方案一：为预处理和后处理步骤各自编写一个MapReduce作业，并将其链接起来。在这些步骤中可以使用IdentityReducer。而在此过程中每一个步骤的中间结果都需要占用I/O和存储资源，所以这种方案是非常低效的，一般情况下Hadoop是不建议使用的。
方案二：自己组合mapper和reducer，也就是说自己写mapper去预先调用所有的预处理步骤，再让reducer调用所有的后处理步骤。在Hadoop中呢，是引入了ChainMapper和ChainReducer类来简化预处理和后处理的构成，生成的作业表达式类似于：MAP+ | REDUCE | MAP+ 通过按照这个顺序来执行执行多个mapper来预处理数据，并在reduce之后可选的地按序执行多个mapper来做数据的后处理。
例如：有4个mapper作业和一个reduce作业，顺序如下：
Map1 | Map2| Reduce | Map3 | Map4
在这个组合中呢，我们就可以把Map2和Reduce来作业MapReduce作业的核心，把Map1作业预处理步骤，把Map3和Map4作业后处理步骤。
[mw_shl_code=java,true]...
conf.setJobName("chain");
conf.setInputFormat(TextInputFormat.class);
conf.setOutputFormat(TextOutputFormat.class);
JobConf mapAConf = new JobConf(false); ...ChainMapper.addMapper(conf, AMap.class, LongWritable.class, Text.class,Text.class, Text.class, true, mapAConf);
JobConf mapBConf = new JobConf(false); ...ChainMapper.addMapper(conf, BMap.class, Text.class, Text.class,LongWritable.class, Text.class, false, mapBConf);
JobConf reduceConf = new JobConf(false); ...ChainReducer.setReducer(conf, XReduce.class, LongWritable.class, Text.class,Text.class, Text.class, true, reduceConf);
ChainReducer.addMapper(conf, CMap.class, Text.class, Text.class,LongWritable.class, Text.class, false, null);
ChainReducer.addMapper(conf, DMap.class, LongWritable.class,Text.class, LongWritable.class, LongWritable.class, true, null);
FileInputFormat.setInputPaths(conf, inDir);FileOutputFormat.setOutputPath(conf, outDir); ...
JobClient jc = new JobClient(conf); RunningJob job =jc.submitJob(conf); ...[/mw_shl_code]
值得注意的是：在ChainMapper.addMapper()方法的签名中：
publicstatic <k1,v1,k2,v2> void addMapper(JobConf job,
Class<? extends Mapper<k1,v1,k2,v2> kclass,
Class<? extends k1> inputKeyClass,
Class<? extends v2> inputValueClass,
Class<? extends k2> outputKeyClass,
Class<? extends v2> outputValueClass,
boolean byValue,
JobConf mapperConf)
第8个参数为boolean byValue，第一个和最后一个是全局和本地jobconf对象，第二个是mapper类，接下来四个mapper使用的类。
byValue参数：true表示值传递，false表示引用传递。
在标准Mapper中，<k2,v2>是采用值传递被洗牌到不同节点上（传递副本），但是目前我们可以将mapper与另一个链接，就在统一个JVM线程执行，就可以采取引用传递。但是一般来说，map1在调用context.write()写出数据后，这些数据是按约定不会更改的。如果引用传递就会破坏约定。但是使用引用传递会提高效率。如果确定数据不会被破坏，可以设置为false，一般安全起见，设置为true即可。。
接下来就来看看具体是怎么是怎么实现的：
首先，本次的小案例操作了两个数据文件，分别是：
input1：
[mw_shl_code=bash,true]2012-3-1 a
2012-3-2 b
2012-3-3 c
2012-3-4 d
2012-3-5 a
2012-3-6 b
2012-3-7 c
2012-3-3 c[/mw_shl_code]
input2：
[mw_shl_code=bash,true]2012-3-1 b
2012-3-2 a
2012-3-3 b
2012-3-4 d
2012-3-5 a
2012-3-6 c
2012-3-7 d
2012-3-3 c[/mw_shl_code]
目标操作实现结果：
[mw_shl_code=bash,true] 2012年3月3日 c
2012年3月7日 c
2012年3月6日 b
2012年3月5日 a
2012年3月4日 d
2012年3月3日 c
2012年3月2日 b
2012年3月1日 a
2012年3月3日 c
2012年3月7日 d
2012年3月6日 c
2012年3月5日 a
2012年3月4日 d
2012年3月3日 b
2012年3月2日 a
2012年3月1日 b[/mw_shl_code]
Mapper1类：
[mw_shl_code=bash,true]package com.demo.mappers;
import java.io.IOException;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
public class Mapper1 extends Mapper<Object,Text,Text,Text> {
//map将输入中的value复制到输出数据的key上，并直接输出
public void map(Text key,Text value,Context context)
throws IOException,InterruptedException{
context.write(value, new Text(""));
}
}
[/mw_shl_code]
Mapper2类：
[mw_shl_code=java,true]package com.demo.mappers;
import java.io.IOException;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Mapper.Context;
public class Mapper2 extends Mapper<Object,Text,Text,Text> {
public void map(Text key,Text value,Context context)
throws IOException,InterruptedException{
context.write(key, new Text(value.toString().split(" ")[0]));
}
}
[/mw_shl_code]
Reduce1类：
[mw_shl_code=java,true]package com.demo.reducers;
import java.io.IOException;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
public class Reducer1 extends Reducer<Text,Text,Text,Text> {
@Override
protected void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, Text>.Context context)
throws IOException, InterruptedException {
for(Text value : values){
String[] s = value.toString().split("-");
StringBuffer disStr = new StringBuffer();
disStr.append(s[0]).append("年").append(s[1]).append("月").append(s[2].split(" ")[0]).append("日").append(" "+s[2].split(" ")[1]);
context.write(key, new Text(disStr.toString()));
}
}
}
[/mw_shl_code]
程序入口：
[mw_shl_code=java,true]package com.demo.test;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapreduce.lib.chain.ChainMapper;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.chain.ChainReducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import com.demo.mappers.Mapper1;
import com.demo.mappers.Mapper2;
import com.demo.reducers.Reducer1;
public class FirstMain {
public static void main(String[] args) {
try {
Configuration conf = new Configuration();
Job job = Job.getInstance(conf, "MyJob");
job.setJarByClass(FirstMain.class);
//设置Map和Reduce处理链
JobConf mapper1Conf = new JobConf(false);
ChainMapper.addMapper(job, Mapper1.class, LongWritable.class, Text.class, Text.class, Text.class, mapper1Conf);
JobConf mapper2Conf = new JobConf(false);
ChainMapper.addMapper(job, Mapper2.class, Text.class, Text.class, Text.class, Text.class, mapper2Conf);
JobConf reduceConf = new JobConf(false);
ChainReducer.setReducer(job, Reducer1.class, LongWritable.class, Text.class, Text.class, Text.class, reduceConf);
//设置输出类型
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
//设置输入和输出目录
FileInputFormat.addInputPath(job, new Path("hdfs://localhost:9000/input/input*"));
FileOutputFormat.setOutputPath(job, new Path("hdfs://localhost:9000/output/firstOutput3"));
System.exit(job.waitForCompletion(true) ? 1 : 0);
} catch (Exception e) {
e.printStackTrace();
}
}
}[/mw_shl_code]
运行结果：
|
/2 
京ICP备2020039040号
简书网举报电话：021-34700000
提升卡
置顶卡
沉默卡
喧嚣卡
变色卡
千斤顶
显身卡
