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

ndlli

业务场景：
假设用户在某处(例如某个网页或者某个地点)的活动会有一个日志，通过日志清洗出用户的一个点击流或者路径流，从而为后续分析做准备。
例子中使用了自定义的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配置。