当MapReduce模型中,reduce执行的任务为统计分类类型的值总量或去重后的数量,或最大值最小值时,可以考虑在Map输出后进行combine操作;这样可以减少网络传输带来的开销,同时减轻了reduce任务的负担。
Combine操作是运行在每个节点上的,只会影响本地Map的输出结果;Combine的输入为本地map的输出结果(一般是数据在溢出到磁盘之前,可以减少IO开销),其输出则作为reduce的输入。
很多时候combine的逻辑和reduce的逻辑是相同的,因此两者可以共用Reducer体;这个时候只需要在客户端中设置Map类之后,Reduce类之前加入一行代码:job.setCombinerClass(MyReducer.class);。如果需要自定义combiner类,可以类似这样(示例为从HBase表读取数据,计算,然后写入另一个HBase表中):
public class Test {
public static class MyMapper extends TableMapper<ImmutableBytesWritable, ImmutableBytesWritable> {
@Override
public void map(ImmutableBytesWritable ibw, Result result, Context context) throws IOException, InterruptedException {
// 从HBase中读取数据后,进行map相关操作
// ...
// 将map结果输出到缓冲区
context.write(new ImmutableBytesWritable(myMapKey, new ImmutableBytesWritable(myMapValue));
}
}
public static class MyCombiner extends TableReducer<ImmutableBytesWritable, ImmutableBytesWritable, ImmutableBytesWritable>{
@Override
public void reduce(ImmutableBytesWritable mapKey, Iterable<ImmutableBytesWritable> mapValues, Context context) throws IOException, InterruptedException {
// 一些操作,如distinct, sum, max, distinct
// ...
// 将combine结果输出到缓冲区
context.write(mapKey, new ImmutableBytesWritable(combineValue));
}
}
public static class MyReducer extends TableReducer<ImmutableBytesWritable, ImmutableBytesWritable, ImmutableBytesWritable> {
@Override
public void reduce(ImmutableBytesWritable reduceKeyIBW, Iterable<ImmutableBytesWritable> reduceValues, Context context) throws IOException, InterruptedException {
// 一些操作,如distinct, sum, max, distinct
// ...
// 将reduce结果(类型Put)输出到缓冲区,用于插入到指定HBase表中
context.write(null, put);
}
}
public static void main(String[] args) throws Exception {
Configuration configuration = HBaseConfiguration.create();
Job job = new Job(configuration, "test_mr");
job.setJarByClass(Test.class);
job.setNumReduceTasks(4); // reduce任务数量默认为1
Scan scan = new Scan();
scan.setCacheBlocks(false);
scan.setCaching(1000); //每次从服务器端读取的行数,默认为配置文件中设置的值
TableMapReduceUtil.initTableMapperJob("table_for_read", scan, MyMapper.class, ImmutableBytesWritable.class, ImmutableBytesWritable.class, job);
job.setCombinerClass(MyCombiner.class); // 设置combiner
TableMapReduceUtil.initTableReducerJob("table_to_write", MyReducer.class, job);
job.waitForCompletion(true);
}
}
当combine操作适用而且Map的输出结果数量很大时,combine的作用是很明显的。
以下是某次测试环境中使用combine前后的对比效果,可以看到Combine output records接近Combine input records的三分之一, 而reduce的输入规模接近原来的十分之一。
两次实际的测试用时分别为50mins16sec, 30mins6sec,耗时节省了接近一半。如图所示:
分享到:
相关推荐
hadoop-MapReduce #Template减少边数据联接欺诈客户检测
修改mapper和reducer数量,如何使用combiner,什么时候该选择哪个writeable等。资料里很详细说明了。
4.5 Hadoop 的Streaming 4.5.1 通过Unix命令使用Streaming 4.5.2 通过脚本使用Streaming 4.5.3 用Streaming处理键/值对 4.5.4 通过Aggregate包使用Streaming 4.6 使用combiner 提升性能 4.7 温故知新 4.8 小...
3 Hadoop MapReduce平台使用 5 3.1 streaming介绍 5 3.2 C语言Map-Reduce程序示例 6 3.2.1计算任务 6 3.2.2 Mapper算法设计 7 3.2.3 Reducer算法设计 8 3.2.4 作业提交命令 9 3.3 shell Map-Reduce程序示例 9 3.3.1...
人工智能-hadoop
基于Hadoop2.2.0的词频统计的例子。包含一个大概十万以上单词的测试数据文件。重写了Partitioner和Combiner,供学习之用。 访问博文 http://blog.csdn.net/zythy/article/details/17852579 以查看详细讲解。
面试hadoop可能被问到的问题你能回答出几个 ? 1、hadoop运行的原理? 2、mapreduce的原理? 3、HDFS存储的机制?...4、举一个简单的例子说明mapreduce是怎么来运行的 ?...5、面试的人给你出一些...6、hadoop中Combiner的作用?
combiner技术点49 超炫的使用比较器的快速排序6.4.4 减轻倾斜技术点50 收集倾斜数据技术点51 减轻reducer 阶段倾斜6.4.5 在MapReduce 中优化用户的Java 代码6.4.6 数据序列化6.5 本章小结第4 部分 ...
单词统计的MapReduce源码,统计多个文本数据集,最终输出每个单词的出现次数,可帮功能扩展修改 Map阶段 采集数据 Combiner阶段 合并数据 Reduce阶段 最终处理,进行排序等自定义操作 每个阶段都会打印对应的数据...
org.apache.hadoop.mapreduce.lib.output.FileOutputFormat; public class MaxTemperature { public static void main(String[] args) throws Exception { if (args.length != 2) { System.err.println("Usage: ...
724.5.4 通过Aggregate包使用Streaming 754.6 使用combiner提升性能 804.7 温故知新 834.8 小结 844.9 更多资源 84第5章 高阶MapReduce 855.1 链接MapReduce作业 855.1.1 顺序链接MapReduce作业 855...
Combiner 就相当于map后多reduce几次。 排序 如果想将mapreduce结果排序,需将排序对象作为键值。 案例:将利润求和后按照顺序排序 数据源 profit.txt 编号 | 姓名 | 收入 | 支出 1 ls 2850 100 2 ls 3566 200 3 ls...
(1)统计全球每年的最高气温和最低气温。 (2)MapReduce输出结果包含年份、最高气温、最低气温,并按最高气温降序排序。如果最高气温相同,则按最低气温升序排序...(5)应用ToolRunner的使用和Eclipse提交MapReduce任务。
编写MapReduce基础程序4.1 获得专利数据集4.1.1 专利引用数据4.1.2 专利描述数据4.2 构建MapReduce 程序的基础模板4.3 计数4.4 适应Hadoop API 的改变4.5 Hadoop 的Streaming4.5.1 通过Unix命令使用Streaming4.5.2 ...
Hadoop 培训课程(4)MapReduce_2 标准和自定义计数器* Combiner和Partitioner编程** 自定义排序和分组编程** 常见的MapReduce算法** ---------------------------加深拓展---------------------- 常见大数据处理方法*
hadoop权威指南第三版(英文版)。 Foreword . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . . xiii Preface . . . . . . ....
技术点48 使用 combiner 技术点49 超炫的使用比较器的快速排序 6.4.4 减轻倾斜 技术点50 收集倾斜数据 技术点51 减轻reducer 阶段倾斜 6.4.5 在MapReduce 中优化用户的Java 代码 6.4.6 数据序列化 6.5...
644.5 Hadoop的Streaming 674.5.1 通过Unix命令使用Streaming 684.5.2 通过脚本使用Streaming 694.5.3 用Streaming处理键/值对 724.5.4 通过Aggregate包使用Streaming 754.6 使用combiner提升性能 804.7 温故知新 ...
(1)统计全球每年的最高气温和最低气温。 (2)MapReduce输出结果包含年份、最高气温、最低气温,并按最高气温降序排序。如果最高气温相同,则按最低气温升序排序...(5)应用ToolRunner的使用和Eclipse提交MapReduce任务。