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

之前的“trackinfo数据清洗”例子中为使用combiner，这个列子通过改写mapper和reducer以支持combiner，同时使用1.75因子计算的reducer task数量。http://gqm.iteye.com/blog/1935541
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阶段的运算以及内存的消耗。