- 浏览: 24256 次
- 性别:
- 来自: 上海
最新评论
业务场景:
假设用户在某处(例如某个网页或者某个地点)的活动会有一个日志,通过日志清洗出用户的一个点击流或者路径流,从而为后续分析做准备。
例子中使用了自定义的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 1000avro非常适合用于hadoop。在开发的时候可能有这样的场景 ... -
[实验]hadoop例子 trackinfo数据清洗的改写
2013-09-03 10:42 1040之前的“trackinfo数据清洗”例子中为使用combine ... -
[笔记]hadoop tutorial - Reducer
2013-09-03 10:15 702引用Reducer reduces a set of inte ... -
[环境] 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. 熟练掌握 Linux 基本命令。 2. 掌握静态 IP 地址的配置、主机名和域名映射的修改。 3. 掌握 Linux 环境下 Java 的安装、环境变量的配置、Java 基本命令的使用。 ...
Hadoop构建数据仓库实践1——王雪迎
基于Hadoop网站流量日志数据分析系统 1、典型的离线流数据分析系统 2、技术分析 - Hadoop - nginx - flume - hive - mysql - springboot + mybatisplus+vcharts nginx + lua 日志文件埋点的 基于Hadoop网站流量...
该文档目录如下: ...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结合起来处理大数据。