- 浏览: 24255 次
- 性别:
- 来自: 上海
最新评论
之前的“trackinfo数据清洗”例子中为使用combiner,这个列子通过改写mapper和reducer以支持combiner,同时使用1.75因子计算的reducer task数量。http://gqm.iteye.com/blog/1935541
Mapper
Reducer
Driver
hadoop job -status job_201308281640_0010
Job: job_201308281640_0010
file: hdfs://node04vm01:9000/tmp/hadoop-hue/mapred/staging/hue/.staging/job_201308281640_0010/job.xml
tracking URL: http://node04vm01:50030/jobdetails.jsp?jobid=job_201308281640_0010
map() completion: 1.0
reduce() completion: 1.0
Counters: 30
Job Counters
Launched reduce tasks=9
SLOTS_MILLIS_MAPS=4936623
Total time spent by all reduces waiting after reserving slots (ms)=0
Total time spent by all maps waiting after reserving slots (ms)=0
Rack-local map tasks=2
Launched map tasks=274
Data-local map tasks=272
SLOTS_MILLIS_REDUCES=4300151
File Output Format Counters
Bytes Written=5875653493
FileSystemCounters
FILE_BYTES_READ=17022188257
HDFS_BYTES_READ=17510078986
FILE_BYTES_WRITTEN=25331743227
HDFS_BYTES_WRITTEN=5875653493
File Input Format Counters
Bytes Read=17510042672
Map-Reduce Framework
Map output materialized bytes=8306340148
Map input records=254655920
Reduce shuffle bytes=8306340148
Spilled Records=357829155
Map output bytes=9004010008
Total committed heap usage (bytes)=56888983552
CPU time spent (ms)=4844340
Combine input records=499067793
SPLIT_RAW_BYTES=36314
Reduce input records=41986484
Reduce input groups=3651914
Combine output records=337948330
Physical memory (bytes) snapshot=71151529984
Reduce output records=3651914
Virtual memory (bytes) snapshot=210540683264
Map output records=203105947
总结
Mapper
public class TrackInfoCleansingMapper extends Mapper<Object, Text, Text, TrackInfoArrayWritable> { private Text user = new Text(); private TrackInfo track = new TrackInfo(); private TrackInfoArrayWritable array = new TrackInfoArrayWritable(); static final int USER_MIN_LEN = 6; @Override protected void map(Object key, Text value, Context context) throws IOException, InterruptedException { StringTokenizer itr = new StringTokenizer(value.toString(), ","); int index = 0; while (itr.hasMoreTokens()) { if (index == 0) { track.getLocation().getMainLoc().set(itr.nextToken()); } else if (index == 1) { track.getLocation().getSubLoc().set(itr.nextToken()); } else if (index == 4) { user.set(itr.nextToken()); if (user.getLength() < USER_MIN_LEN) { // illegal user, skip line break; } } else if (index == 6) { track.getTrackTime().set(itr.nextToken()); array.set(new TrackInfo[] { track }); context.write(user, array); // the map intermediate data is OK, skip other info break; } else { itr.nextToken(); } index++; } } }
Reducer
public class TrackInfoCleansingReducer extends Reducer<Text, TrackInfoArrayWritable, Text, TrackInfoArrayWritable> { private TrackInfoArrayWritable tracks = new TrackInfoArrayWritable(); private List<TrackInfo> rentList = new ArrayList<>(); @Override protected void reduce(Text key, Iterable<TrackInfoArrayWritable> values, Context context) throws IOException, InterruptedException { int index = 0; List<TrackInfo> list = new LinkedList<>(); TrackInfo rent = null; TrackInfo info = null; for (TrackInfoArrayWritable array : values) { for (Writable item : array.get()) { info = (TrackInfo) item; // if rentList has item, then use it, // otherwise create a new item to use and add it to the // rentList. if (index < rentList.size()) { rent = rentList.get(index); } else { // new instance rent = new TrackInfo(); rentList.add(rent); } index++; // copy info to rent rent.getTrackTime().set(info.getTrackTime().toString()); rent.getLocation().getMainLoc() .set(info.getLocation().getMainLoc().toString()); rent.getLocation().getSubLoc() .set(info.getLocation().getSubLoc().toString()); list.add(rent); } } Collections.sort(list, new Comparator<TrackInfo>() { @Override public int compare(TrackInfo o1, TrackInfo o2) { return o1.compareTo(o2); } }); TrackInfo[] temp = new TrackInfo[list.size()]; list.toArray(temp); tracks.set(temp); context.write(key, tracks); } }
Driver
public class TrackInfoCleansing extends Configured implements Tool { public static void main(String[] args) throws Exception { int exitCode = ToolRunner.run(new TrackInfoCleansing(), args); System.exit(exitCode); } @Override public int run(String[] args) throws Exception { if(args.length != 2){ System.out.printf("Usage %s [generic options] <in> <out>\n", getClass().getName()); ToolRunner.printGenericCommandUsage(System.out); return -1; } Configuration conf = new Configuration(); conf.set("fs.default.name", "hdfs://node04vm01:9000"); Job job = new Job(conf, "track info cleansing"); job.setNumReduceTasks(7); job.setJarByClass(TrackInfoCleansing.class); job.setMapperClass(TrackInfoCleansingMapper.class); job.setCombinerClass(TrackInfoCleansingReducer.class); job.setReducerClass(TrackInfoCleansingReducer.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(TrackInfoArrayWritable.class); job.setOutputKeyClass(Text.class); job.setOutputValueClass(TrackInfoArrayWritable.class); job.setOutputFormatClass(SequenceFileOutputFormat.class); FileInputFormat.setInputPaths(job, new Path(args[0])); FileOutputFormat.setOutputPath(job, new Path(args[1])); return job.waitForCompletion(true) ? 0 : 1; } }
hadoop job -status job_201308281640_0010
Job: job_201308281640_0010
file: hdfs://node04vm01:9000/tmp/hadoop-hue/mapred/staging/hue/.staging/job_201308281640_0010/job.xml
tracking URL: http://node04vm01:50030/jobdetails.jsp?jobid=job_201308281640_0010
map() completion: 1.0
reduce() completion: 1.0
Counters: 30
Job Counters
Launched reduce tasks=9
SLOTS_MILLIS_MAPS=4936623
Total time spent by all reduces waiting after reserving slots (ms)=0
Total time spent by all maps waiting after reserving slots (ms)=0
Rack-local map tasks=2
Launched map tasks=274
Data-local map tasks=272
SLOTS_MILLIS_REDUCES=4300151
File Output Format Counters
Bytes Written=5875653493
FileSystemCounters
FILE_BYTES_READ=17022188257
HDFS_BYTES_READ=17510078986
FILE_BYTES_WRITTEN=25331743227
HDFS_BYTES_WRITTEN=5875653493
File Input Format Counters
Bytes Read=17510042672
Map-Reduce Framework
Map output materialized bytes=8306340148
Map input records=254655920
Reduce shuffle bytes=8306340148
Spilled Records=357829155
Map output bytes=9004010008
Total committed heap usage (bytes)=56888983552
CPU time spent (ms)=4844340
Combine input records=499067793
SPLIT_RAW_BYTES=36314
Reduce input records=41986484
Reduce input groups=3651914
Combine output records=337948330
Physical memory (bytes) snapshot=71151529984
Reduce output records=3651914
Virtual memory (bytes) snapshot=210540683264
Map output records=203105947
总结
- 使用Combiner对HDFS度读写是一样的,说明并不影响结果。
- 使用Combiner可以减少本地FS的IO,即减少mapper阶段的中间结果的FS的IO。
- 使用Combiner在减少中间结果的IO的过程也减少了Reducer的shuffle阶段network io,即copy的数量,也减少了reducer input records的量。
- 使用Combiner增加了mapper阶段的运算以及内存的消耗。
发表评论
-
[实验]avro与non-avro的mapred例子-wordcount改写
2013-09-03 16:15 1000avro非常适合用于hadoop。在开发的时候可能有这样的场景 ... -
[笔记]hadoop tutorial - Reducer
2013-09-03 10:15 702引用Reducer reduces a set of inte ... -
[实验]hadoop例子 trackinfo数据清洗
2013-09-02 17:24 2528业务场景: 假设用户在某处(例如某个网页或者某个地点)的活动会 ... -
[环境] hadoop 开发环境maven管理
2013-09-02 17:02 1431贴一下整理的maven管理配置(待补充) <proj ... -
[笔记]avro 介绍及官网例子
2013-09-02 14:22 3838Apache Avro是一个独立于编程语言的数据序列化系统。旨 ... -
[实验]hadoop例子 在线用户分析
2013-08-30 15:54 857一个简单的业务场景和例子。由wordcount例子改写。 业 ... -
[笔记]hadoop mapred InputFormat分析
2013-08-30 13:43 1220Hadoop MapReduce的编程接口层主要有5个可编程组 ... -
[笔记]hdfs namenode FSNamesystem分析
2013-08-30 09:18 1114NameNode在内存中维护整个文件系统的元数据镜像,用于HD ... -
[笔记]hdfs namenode FSImage分析1
2013-08-29 15:10 1832元数据文件fsimage的分析 fsimage为元数据镜像文件 ... -
[实验]集群hadoop配置
2013-08-28 16:53 812环境 hadoop1.2.0 CentOS release ... -
[实验]单机hadoop配置
2013-08-28 14:16 572环境: hadoop1.2.0 配置 修改conf/core ... -
[问题解决]hadoop eclipse plugin
2013-08-27 09:22 936环境: hadoop 1.2.0 问题: eclipse报错& ...
相关推荐
针对本次实验,我们需要用到Hadoop集群作为模拟大数据的分析软件,集群环境必须要包括,hdfs,hbase,hive,flume,sqoop等插件,最后结合分析出来的数据进行可视化展示,需要用到Python(爬取数据集,可视化展示)...
基于Hadoop的全国酒店数据清洗项目源码+报告.zip基于Hadoop的全国酒店数据清洗项目源码+报告.zip基于Hadoop的全国酒店数据清洗项目源码+报告.zip基于Hadoop的全国酒店数据清洗项目源码+报告.zip基于Hadoop的全国酒店...
Hadoop豆瓣电影数据分析(Hadoop)操作源码
eclipse hadoop例子源代码 eclipse hadoop例子源代码
可以作为大数据预处理的MapReduce代码的参考!!! -执行脚本文件: cd /home/hadoop/logfiles/ source format_run_2013_o5_30.sh source format_run_2013_o5_31.sh 执行我们的脚本文件,可以用source或者./
Hadoop分布式数据清洗方案一种基于孤立点挖掘的Hadoop数据清洗算法的研究.pptx
Hadoop分析气象数据完整版源代码(含Hadoop的MapReduce代码和SSM框架) 《分布式》布置了一道小作业,这是作业的所有代码,里面包含了Hadoop的MapReduce代码、和SSM框架显示数据的代码
2022毕业设计,基于 Hadoop 的游戏数据分析系统源码.zip
是大数据课程大作业,基于Hadoop的电影影评数据分析,需要安装Hadoop,了解MapReduce 和HDFS。
这里提出了一种基于分布式计算技术进行管理和存储海量海洋科学数据方法,构建了海量海洋科学数据存储平台解决方案,采用Linux集群技术,设计开发一个基于Hadoop的海量数据存储平台.系统由五大模块组成,有系统管理模块、...
毕业设计,基于 Hadoop 的游戏数据分析系统毕业设计,基于 Hadoop 的游戏数据分析系统毕业设计,基于 Hadoop 的游戏数据分析系统毕业设计,基于 Hadoop 的游戏数据分析系统毕业设计,基于 Hadoop 的游戏数据分析系统...
基于Hadoop实现的数据云盘系统项目源码资料大全.zip基于Hadoop实现的数据云盘系统,实现不同用户有不同的存储空间,不同的用户身份有不同的权限操作等功能软件设计思路 1、用户登录模块设计 输入图片说明 2、检索...
hadoop构建数据仓库实践
第四章(Hadoop大数据处理实战)Hadoop分布式文件系统.pdf第四章(Hadoop大数据处理实战)Hadoop分布式文件系统.pdf第四章(Hadoop大数据处理实战)Hadoop分布式文件系统.pdf第四章(Hadoop大数据处理实战)Hadoop分布式文件...
基于Hadoop网站流量日志数据分析系统 1、典型的离线流数据分析系统 2、技术分析 - Hadoop - nginx - flume - hive - mysql - springboot + mybatisplus+vcharts nginx + lua 日志文件埋点的 基于Hadoop网站流量...
部署全分布模式Hadoop集群 实验报告一、实验目的 1. 熟练掌握 Linux 基本命令。 2. 掌握静态 IP 地址的配置、主机名和域名映射的修改。 3. 掌握 Linux 环境下 Java 的安装、环境变量的配置、Java 基本命令的使用。 ...
Hadoop构建数据仓库实践1——王雪迎
该文档目录如下: ...1.1 基于Hadoop的数据仓库Hive学习指南 1.2实验环境 1.3实验原理 1.3.1 Hive简介 1.3.2 Hive安装 1.3.3安装并配置mysql 1.3.5 Hive简单编程实践 1.3.4 Hive的常用HiveQL操作
数据分析R,英文版。主要说明R和hadoop结合起来处理大数据。