[実験]hadoop例trocinfoデータ洗浄の書き換え

以前の「trocinfoデータ洗浄」の例では、combinerを使用していましたが、この列はmapperとreducerを書き換えて、cobinerをサポートし、同時に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>
", 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-hu/mapred/staging/hu/staging/job_201308281640_0010/job.xml
trocing URL：http://node04vm01:50030/jobdetails.jsp?jobid=job_201308281640_0010
map()complettetion:1.0
reduce()complection:1.0
Counters:30
Job Counters
Launched reduce taskys=9
SLOTS_MILLIS_MAPS=4936623
Total time spent by all reduces waiting after reving slaots(ms)=0
Total time spent by all maps waiting after reverving slaots(ms)=0
Rack-local map taskys=2
Launched map task=274
Data-local map task=272
SLOTS_MILLIS_REDUCES=430051
File Output Format Counters
Bytes Written=5875653493
FileSystem Counters
FILE_BYTES_READ=1702216257
HFS_BYTES_READ=17510078986
FILE_BYTES_WRITTEN=25331743227
HFS_BYTES_WRITTEN=5875653493
File Input Format Counters
Bytes Read=17510042672
Map-Reduce Frame ework
Map output materialized bytes=830640148
Map input records=254655920
Reduce shuffle bytes=830640148
Spilled Records=357829155
Map output bytes=900401008
Total comitted heap usage(bytes)=56888983552
CPU time spent(ms)=4844340
Compinput records=499067793
SPLIT_RAW_BYTES=36314
Reduce input records=41986484
Reduce input groups=3651914
Compone output records=337948330
Physical memory(bytes)snapshot=711529984
Reduce output records=3651914
Virtual memory(bytes)snapshot=21054063264
Map output records=2010947
締め括りをつける

Cobinerを使用してHFS度の読み書きは同じで、結果に影響しないと説明しています。

は、Cobinerを使用して、ローカルFSのIO、すなわち、mapper段階の中間結果を低減するFSのIOを低減することができる。

Cobinerを使用して中間結果を低減するIOのプロセスもReducerのshuffle段階network io、すなわちcopyの数を減らし、reducer input recordsの量を減らした。

は、Commbinerを使用して、mapper段階の演算およびメモリの消費を増加させた。