- 浏览: 24834 次
- 性别:
- 来自: 上海
最新评论
业务场景:
假设用户在某处(例如某个网页或者某个地点)的活动会有一个日志,通过日志清洗出用户的一个点击流或者路径流,从而为后续分析做准备。
例子中使用了自定义的Hadoop的Writable类
位置类Location.java
定义了主位置信息mainLoc和细分位置信息subLoc
带时间的位置类TrackInfo.java
定义了时间信息trackTime和位置信息location,由于期望在输出的时候对TrackInfo根据时间排序,故实现了WritableComparable接口。
TrackInfo数组类TrackInfoArrayWritable.java
Mapper
对日志信息识别出用户和位置以及发生的时间
Reducer
对用户的TrackInfo按时间序写出
Driver类
hadoop job -status job_201308281640_0008
Job: job_201308281640_0008
file: hdfs://node04vm01:9000/tmp/hadoop-hue/mapred/staging/hue/.staging/job_201308281640_0008/job.xml
tracking URL: http://node04vm01:50030/jobdetails.jsp?jobid=job_201308281640_0008
map() completion: 1.0
reduce() completion: 1.0
Counters: 30
Job Counters
Launched reduce tasks=5
SLOTS_MILLIS_MAPS=3610050
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=1
Launched map tasks=275
Data-local map tasks=274
SLOTS_MILLIS_REDUCES=2304285
File Output Format Counters
Bytes Written=5875655704
FileSystemCounters
FILE_BYTES_READ=22615983064
HDFS_BYTES_READ=17510078986
FILE_BYTES_WRITTEN=31223474658
HDFS_BYTES_WRITTEN=5875655704
File Input Format Counters
Bytes Read=17510042672
Map-Reduce Framework
Map output materialized bytes=8597804618
Map input records=254655920
Reduce shuffle bytes=8597804618
Spilled Records=737245107
Map output bytes=8191586220
Total committed heap usage (bytes)=55739351040
CPU time spent (ms)=3336100
Combine input records=0
SPLIT_RAW_BYTES=36314
Reduce input records=203105947
Reduce input groups=3651914
Combine output records=0
Physical memory (bytes) snapshot=69177683968
Reduce output records=3651914
Virtual memory (bytes) snapshot=208306130944
Map output records=203105947
问题总结
假设用户在某处(例如某个网页或者某个地点)的活动会有一个日志,通过日志清洗出用户的一个点击流或者路径流,从而为后续分析做准备。
例子中使用了自定义的Hadoop的Writable类
位置类Location.java
定义了主位置信息mainLoc和细分位置信息subLoc
public class Location implements Writable { private final Text mainLoc; private final Text subLoc; public Location() { this.mainLoc = new Text(); this.subLoc = new Text(); } public Text getMainLoc() { return mainLoc; } public Text getSubLoc() { return subLoc; } @Override public void write(DataOutput out) throws IOException { mainLoc.write(out); subLoc.write(out); } @Override public void readFields(DataInput in) throws IOException { mainLoc.readFields(in); subLoc.readFields(in); } }
带时间的位置类TrackInfo.java
定义了时间信息trackTime和位置信息location,由于期望在输出的时候对TrackInfo根据时间排序,故实现了WritableComparable接口。
public class TrackInfo extends BinaryComparable implements WritableComparable<BinaryComparable> { private final Text trackTime; private final Location location; public TrackInfo() { this.trackTime = new Text(); this.location = new Location(); } @Override public void write(DataOutput out) throws IOException { trackTime.write(out); location.write(out); } @Override public void readFields(DataInput in) throws IOException { trackTime.readFields(in); location.readFields(in); } @Override public int getLength() { return trackTime.getLength(); } @Override public byte[] getBytes() { return trackTime.getBytes(); } public Text getTrackTime() { return trackTime; } public Location getLocation() { return location; } }
TrackInfo数组类TrackInfoArrayWritable.java
public class TrackInfoArrayWritable extends ArrayWritable { public TrackInfoArrayWritable() { super(TrackInfo.class); } }
Mapper
对日志信息识别出用户和位置以及发生的时间
public class TrackInfoCleansingMapper extends Mapper<Object, Text, Text, TrackInfo> { private Text user = new Text(); private TrackInfo track = new TrackInfo(); 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()); context.write(user, track); // the map intermediate data is OK, skip other info break; } else { itr.nextToken(); } index++; } } }
Reducer
对用户的TrackInfo按时间序写出
public class TrackInfoCleansingReducer extends Reducer<Text, TrackInfo, Text, TrackInfoArrayWritable> { private TrackInfoArrayWritable tracks = new TrackInfoArrayWritable(); private List<TrackInfo> rentList = new ArrayList<>(); @Override protected void reduce(Text key, Iterable<TrackInfo> values, Context context) throws IOException, InterruptedException { int index = 0; List<TrackInfo> list = new LinkedList<>(); TrackInfo rent = null; for (TrackInfo info : values) { // 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(4); job.setJarByClass(TrackInfoCleansing.class); job.setMapperClass(TrackInfoCleansingMapper.class); job.setReducerClass(TrackInfoCleansingReducer.class); job.setMapOutputKeyClass(Text.class); job.setMapOutputValueClass(TrackInfo.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_0008
Job: job_201308281640_0008
file: hdfs://node04vm01:9000/tmp/hadoop-hue/mapred/staging/hue/.staging/job_201308281640_0008/job.xml
tracking URL: http://node04vm01:50030/jobdetails.jsp?jobid=job_201308281640_0008
map() completion: 1.0
reduce() completion: 1.0
Counters: 30
Job Counters
Launched reduce tasks=5
SLOTS_MILLIS_MAPS=3610050
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=1
Launched map tasks=275
Data-local map tasks=274
SLOTS_MILLIS_REDUCES=2304285
File Output Format Counters
Bytes Written=5875655704
FileSystemCounters
FILE_BYTES_READ=22615983064
HDFS_BYTES_READ=17510078986
FILE_BYTES_WRITTEN=31223474658
HDFS_BYTES_WRITTEN=5875655704
File Input Format Counters
Bytes Read=17510042672
Map-Reduce Framework
Map output materialized bytes=8597804618
Map input records=254655920
Reduce shuffle bytes=8597804618
Spilled Records=737245107
Map output bytes=8191586220
Total committed heap usage (bytes)=55739351040
CPU time spent (ms)=3336100
Combine input records=0
SPLIT_RAW_BYTES=36314
Reduce input records=203105947
Reduce input groups=3651914
Combine output records=0
Physical memory (bytes) snapshot=69177683968
Reduce output records=3651914
Virtual memory (bytes) snapshot=208306130944
Map output records=203105947
问题总结
- Writable方式序列化的输出数据不直观,需要用Writable反序列化才能看到实际数据。(可用作中间结果的序列化框架,如果有其他用途考虑改用其他框架,例如avro)
- 运行时需要考虑并设置reducer的数量。
- 需要考虑使用Combiner以减少reduce input records的数量。
- 为防止运行中出现了java heap space的OOM,需要调优程序(如设置combiner,优化排序等)和JVM配置。
发表评论
-
[实验]avro与non-avro的mapred例子-wordcount改写
2013-09-03 16:15 1045avro非常适合用于hadoop。在开发的时候可能有这样的场景 ... -
[实验]hadoop例子 trackinfo数据清洗的改写
2013-09-03 10:42 1082之前的“trackinfo数据清洗”例子中为使用combine ... -
[笔记]hadoop tutorial - Reducer
2013-09-03 10:15 732引用Reducer reduces a set of inte ... -
[环境] hadoop 开发环境maven管理
2013-09-02 17:02 1463贴一下整理的maven管理配置(待补充) <proj ... -
[笔记]avro 介绍及官网例子
2013-09-02 14:22 3883Apache Avro是一个独立于编程语言的数据序列化系统。旨 ... -
[实验]hadoop例子 在线用户分析
2013-08-30 15:54 898一个简单的业务场景和例子。由wordcount例子改写。 业 ... -
[笔记]hadoop mapred InputFormat分析
2013-08-30 13:43 1264Hadoop MapReduce的编程接口层主要有5个可编程组 ... -
[笔记]hdfs namenode FSNamesystem分析
2013-08-30 09:18 1147NameNode在内存中维护整个文件系统的元数据镜像,用于HD ... -
[笔记]hdfs namenode FSImage分析1
2013-08-29 15:10 1874元数据文件fsimage的分析 fsimage为元数据镜像文件 ... -
[实验]集群hadoop配置
2013-08-28 16:53 853环境 hadoop1.2.0 CentOS release ... -
[实验]单机hadoop配置
2013-08-28 14:16 606环境: hadoop1.2.0 配置 修改conf/core ... -
[问题解决]hadoop eclipse plugin
2013-08-27 09:22 979环境: hadoop 1.2.0 问题: eclipse报错& ...
相关推荐
【基于Hadoop豆瓣电影数据分析实验报告】 在大数据时代,对海量信息进行高效处理和分析是企业决策的关键。Hadoop作为一款强大的分布式计算框架,自2006年诞生以来,已经在多个领域展现了其卓越的数据处理能力。本...
基于孤立点挖掘的Hadoop数据清洗算法研究 基于孤立点挖掘的Hadoop数据清洗算法是指使用Hadoop分布式计算平台,结合孤立点挖掘技术,实现大规模数据的清洗和质量提高。该算法可以对不同类型的脏数据进行有效清洗,...
基于Hadoop的全国酒店数据清洗项目源码+报告.zip基于Hadoop的全国酒店数据清洗项目源码+报告.zip基于Hadoop的全国酒店数据清洗项目源码+报告.zip基于Hadoop的全国酒店数据清洗项目源码+报告.zip基于Hadoop的全国酒店...
Hadoop豆瓣电影数据分析(Hadoop)操作源码
【标题】:“Eclipse Hadoop 例子源代码” 在大数据处理领域,Hadoop是一个不可或缺的开源框架,它提供了分布式存储和计算的能力。Eclipse作为Java开发的主流集成开发环境(IDE),也是编写和调试Hadoop程序的重要...
【标题】:“2022毕业设计,基于 Hadoop 的游戏数据分析系统源码” 这个毕业设计项目主要聚焦于使用Hadoop框架开发一个游戏数据分析系统。Hadoop是Apache软件基金会的一个开源分布式计算平台,专为处理和存储大规模...
【基于Hadoop的电影影评数据分析】是一项大数据课程的大作业,旨在利用Hadoop的分布式处理能力来分析电影影评数据。Hadoop是一个由Apache软件基金会开发的开源框架,专为处理和存储大规模数据而设计。它由四个核心...
可以作为大数据预处理的MapReduce代码的参考!!! -执行脚本文件: cd /home/hadoop/logfiles/ source format_run_2013_o5_30.sh source format_run_2013_o5_31.sh 执行我们的脚本文件,可以用source或者./
在这个"MapReduce项目 数据清洗"中,我们将探讨如何使用MapReduce对遗传关系族谱数据进行处理和清洗,以便进行后续分析。 1. **Map阶段**: 在Map阶段,原始数据被分割成多个小块(split),然后分配到不同的工作...
Hadoop可以通过Pig、Hive或Spark SQL等工具进行数据转换和清洗。 3. 数据存储:Hadoop支持多种数据存储格式,如HBase(NoSQL数据库)、Hive(数据仓库工具)和Oozie(工作流调度系统)。根据业务需求,可以选择合适...
Hadoop分析气象数据完整版源代码(含Hadoop的MapReduce代码和SSM框架) 《分布式》布置了一道小作业,这是作业的所有代码,里面包含了Hadoop的MapReduce代码、和SSM框架显示数据的代码
在IT行业中,大数据处理是一项至关重要的任务,而Hadoop作为开源的大数据处理框架,因其高效、可扩展的特性,被广泛应用于气象数据分析等场景。在这个项目中,我们重点关注的是一套完整的Hadoop分析气象数据的代码,...
在大数据处理领域,Hadoop是一个不可或缺的开源框架,它为海量数据的存储和处理提供了高效、可靠的解决方案。本文将深入探讨“Hadoop之外卖订单数据分析系统”,并介绍如何利用Hadoop进行大规模数据处理,以及如何将...
使用hadoop进行数据分析天气数据分析.zip使用hadoop进行数据分析天气数据分析.zip使用hadoop进行数据分析天气数据分析.zip使用hadoop进行数据分析天气数据分析.zip使用hadoop进行数据分析天气数据分析.zip使用hadoop...
【标题】:“基于Hadoop的数据仓库Hive学习指南” 【描述】:该文档是一份针对Hive的学习资料,旨在引导读者理解如何在Hadoop平台上利用Hive进行数据仓库操作和编程实践。它涵盖了Hive的基本概念、安装步骤、实验...
毕业设计,基于 Hadoop 的游戏数据分析系统毕业设计,基于 Hadoop 的游戏数据分析系统毕业设计,基于 Hadoop 的游戏数据分析系统毕业设计,基于 Hadoop 的游戏数据分析系统毕业设计,基于 Hadoop 的游戏数据分析系统...