【转】ChainMapper 实例理解一
2015-12-24 18:09
197 查看
通过ChainMapper可以将多个map类合并成一个map任务。
下面个这个例子没什么实际意思,但是很好的演示了ChainMapper的作用。
源文件
100 tom 90
101 mary 85
102 kate 60
map00的结果,过滤掉100的记录
101 mary 85
102 kate 60
map01的结果,过滤掉101的记录
102 kate 60
reduce结果
102 kate 60
总结:
1.一句话:ChainMapper即在Reduce之前进行多次Mapper
2.ChainMapper必须保证所有的子mapper输入输出是一致的!
3.ChainMapper中的子mapper是线性执行的
下面个这个例子没什么实际意思,但是很好的演示了ChainMapper的作用。
源文件
100 tom 90
101 mary 85
102 kate 60
map00的结果,过滤掉100的记录
101 mary 85
102 kate 60
map01的结果,过滤掉101的记录
102 kate 60
reduce结果
102 kate 60
import java.io.IOException; import java.util.*; import java.lang.String; import org.apache.hadoop.fs.Path; import org.apache.hadoop.conf.*; import org.apache.hadoop.io.*; import org.apache.hadoop.mapred.*; import org.apache.hadoop.util.*; import org.apache.hadoop.mapred.lib.*; public class WordCount { public static class Map00 extends MapReduceBase implements Mapper { public void map(Text key, Text value, OutputCollector output, Reporter reporter) throws IOException { Text ft = new Text(“100″); if(!key.equals(ft)) { output.collect(key, value); } } } public static class Map01 extends MapReduceBase implements Mapper { public void map(Text key, Text value, OutputCollector output, Reporter reporter) throws IOException { Text ft = new Text(“101″); if(!key.equals(ft)) { output.collect(key, value); } } } public static class Reduce extends MapReduceBase implements Reducer { public void reduce(Text key, Iterator values, OutputCollector output, Reporter reporter) throws IOException { while(values.hasNext()) { output.collect(key, values.next()); } } } public static void main(String[] args) throws Exception { JobConf conf = new JobConf(WordCount.class); conf.setJobName(“wordcount00″); conf.setInputFormat(KeyValueTextInputFormat.class); conf.setOutputFormat(TextOutputFormat.class); ChainMapper cm = new ChainMapper(); JobConf mapAConf = new JobConf(false); cm.addMapper(conf, Map00.class, Text.class, Text.class, Text.class, Text.class, true, mapAConf); JobConf mapBConf = new JobConf(false); cm.addMapper(conf, Map01.class, Text.class, Text.class, Text.class, Text.class, true, mapBConf); conf.setReducerClass(Reduce.class); conf00.setOutputKeyClass(Text.class); conf00.setOutputValueClass(Text.class); FileInputFormat.setInputPaths(conf, new Path(args[0])); FileOutputFormat.setOutputPath(conf, new Path(args[1])); JobClient.runJob(conf); } }
总结:
1.一句话:ChainMapper即在Reduce之前进行多次Mapper
2.ChainMapper必须保证所有的子mapper输入输出是一致的!
3.ChainMapper中的子mapper是线性执行的
相关文章推荐
- Android MVVM(一) 介绍
- android源码使用proguard混淆编译及错误总结
- ios中round/ceil/floorf函数的使用
- 140,Swift的字符串
- Android/iOS 和js的交互的兼容性写法
- Objective-C Runtime 总结:消息机制 篇
- Unity和Lua交互
- iOS_获取通讯录
- QWebView下载文件,QUrl中解析文件名
- 微信公共平台php收不到信息解决方法
- android内部空间写文档
- Android 毛玻璃效果弹出菜单动画
- iOS 绘图机制简介,Quartz 2D绘图用CGContextRef绘制音频波形图 —— HERO博客
- android偏门知识小节
- iOS学习之KVO的使用
- 【cocos2d-x】之添加虚拟摇杆
- 安卓WebView与ScrollView嵌套大面积空白解决小技巧和适配
- IOS系列教程
- iOS开发中怎么抓包
- iOS在AppDelegate类中获取其他类中属性的值