`
GQM
  • 浏览: 24870 次
  • 性别: Icon_minigender_1
  • 来自: 上海
社区版块
存档分类
最新评论

[实验]hadoop例子 trackinfo数据清洗

阅读更多
业务场景:
假设用户在某处(例如某个网页或者某个地点)的活动会有一个日志,通过日志清洗出用户的一个点击流或者路径流,从而为后续分析做准备。

例子中使用了自定义的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配置。
分享到:
评论

相关推荐

    基于Hadoop豆瓣电影数据分析实验报告

    【基于Hadoop豆瓣电影数据分析实验报告】 在大数据时代,对海量信息进行高效处理和分析是企业决策的关键。Hadoop作为一款强大的分布式计算框架,自2006年诞生以来,已经在多个领域展现了其卓越的数据处理能力。本...

    Hadoop分布式数据清洗方案一种基于孤立点挖掘的Hadoop数据清洗算法的研究.pptx

    基于孤立点挖掘的Hadoop数据清洗算法研究 基于孤立点挖掘的Hadoop数据清洗算法是指使用Hadoop分布式计算平台,结合孤立点挖掘技术,实现大规模数据的清洗和质量提高。该算法可以对不同类型的脏数据进行有效清洗,...

    基于Hadoop的全国酒店数据清洗项目源码+报告.zip

    基于Hadoop的全国酒店数据清洗项目源码+报告.zip基于Hadoop的全国酒店数据清洗项目源码+报告.zip基于Hadoop的全国酒店数据清洗项目源码+报告.zip基于Hadoop的全国酒店数据清洗项目源码+报告.zip基于Hadoop的全国酒店...

    Hadoop豆瓣电影数据分析(Hadoop)操作源码

    Hadoop豆瓣电影数据分析(Hadoop)操作源码

    eclipse hadoop 例子源代码

    【标题】:“Eclipse Hadoop 例子源代码” 在大数据处理领域,Hadoop是一个不可或缺的开源框架,它提供了分布式存储和计算的能力。Eclipse作为Java开发的主流集成开发环境(IDE),也是编写和调试Hadoop程序的重要...

    2022毕业设计,基于 Hadoop 的游戏数据分析系统源码.zip

    【标题】:“2022毕业设计,基于 Hadoop 的游戏数据分析系统源码” 这个毕业设计项目主要聚焦于使用Hadoop框架开发一个游戏数据分析系统。Hadoop是Apache软件基金会的一个开源分布式计算平台,专为处理和存储大规模...

    基于Hadoop的电影影评数据分析

    【基于Hadoop的电影影评数据分析】是一项大数据课程的大作业,旨在利用Hadoop的分布式处理能力来分析电影影评数据。Hadoop是一个由Apache软件基金会开发的开源框架,专为处理和存储大规模数据而设计。它由四个核心...

    hadoop实训课数据清洗py脚本(MapReduce python代码,可执行文件脚本,使用方法)

    可以作为大数据预处理的MapReduce代码的参考!!! -执行脚本文件: cd /home/hadoop/logfiles/ source format_run_2013_o5_30.sh source format_run_2013_o5_31.sh 执行我们的脚本文件,可以用source或者./

    mapreduce项目 数据清洗

    在这个"MapReduce项目 数据清洗"中,我们将探讨如何使用MapReduce对遗传关系族谱数据进行处理和清洗,以便进行后续分析。 1. **Map阶段**: 在Map阶段,原始数据被分割成多个小块(split),然后分配到不同的工作...

    Hadoop构建数据仓库实践1_hadoop_

    Hadoop可以通过Pig、Hive或Spark SQL等工具进行数据转换和清洗。 3. 数据存储:Hadoop支持多种数据存储格式,如HBase(NoSQL数据库)、Hive(数据仓库工具)和Oozie(工作流调度系统)。根据业务需求,可以选择合适...

    Hadoop分析气象数据完整版源代码(含Hadoop的MapReduce代码和SSM框架)

    Hadoop分析气象数据完整版源代码(含Hadoop的MapReduce代码和SSM框架) 《分布式》布置了一道小作业,这是作业的所有代码,里面包含了Hadoop的MapReduce代码、和SSM框架显示数据的代码

    Hadoop分析气象数据完整版代码

    在IT行业中,大数据处理是一项至关重要的任务,而Hadoop作为开源的大数据处理框架,因其高效、可扩展的特性,被广泛应用于气象数据分析等场景。在这个项目中,我们重点关注的是一套完整的Hadoop分析气象数据的代码,...

    Hadoop之外卖订单数据分析系统

    在大数据处理领域,Hadoop是一个不可或缺的开源框架,它为海量数据的存储和处理提供了高效、可靠的解决方案。本文将深入探讨“Hadoop之外卖订单数据分析系统”,并介绍如何利用Hadoop进行大规模数据处理,以及如何将...

    使用hadoop进行天气数据分析.zip

    使用hadoop进行数据分析天气数据分析.zip使用hadoop进行数据分析天气数据分析.zip使用hadoop进行数据分析天气数据分析.zip使用hadoop进行数据分析天气数据分析.zip使用hadoop进行数据分析天气数据分析.zip使用hadoop...

    基于Hadoop的数据仓库Hive学习指南.doc

    【标题】:“基于Hadoop的数据仓库Hive学习指南” 【描述】:该文档是一份针对Hive的学习资料,旨在引导读者理解如何在Hadoop平台上利用Hive进行数据仓库操作和编程实践。它涵盖了Hive的基本概念、安装步骤、实验...

    毕业设计,基于 Hadoop 的游戏数据分析系统

    毕业设计,基于 Hadoop 的游戏数据分析系统毕业设计,基于 Hadoop 的游戏数据分析系统毕业设计,基于 Hadoop 的游戏数据分析系统毕业设计,基于 Hadoop 的游戏数据分析系统毕业设计,基于 Hadoop 的游戏数据分析系统...

Global site tag (gtag.js) - Google Analytics