HADOOP JAVA MapReduce API

gaojinguan

　　下面结合具体业务讲解一下 HADOOP JAVA MapReduce API。
1.日志格式
　　time=2011-07-05 21:59:56`pid=52249`channelid=3`pos=1`adver=`ex=`monitoring=`guid=`ip=`sn=`dn=924779104-e8f54d2f`bid=`pfid=69`width=240`height=320`ss=240×320`fr=dwjava`ua=`imei=`ln=zh_cn`ext=a`li=g9eKibG8rMqN2tOHt76sy4/a04CyvqGe1dqA17u9p5PWiOm+`gi=`isp=移动`prov=四川`city=成都`ver=7.8.0.87`ver3=7.8.0`poid=1
……
　　多行的key=value日志用 ` 分割
2.统计业务需求
　　
统计各个pid出现的次数，其实也可以结合其他维度统计pid出现的次数，为了便于理解，我只讲最简单的业务需求
3.代码分析
　　mapper中实现两个内置方法，用来处理日志：
　　// 切分日志
public String[] cutLog(String string) {
String[] x = string.split(“`”);
return x;
}
　　// 切分key value
public String[] cutKeyValue(String string) {
String[] x = string.split(“=”);
return x;
}
　　具体调用如下：
　　public void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
NewPidCountMapper mapper = new NewPidCountMapper();
String line = value.toString();
　　String[] x = mapper.cutLog(line.trim());  //先按照`切分
Map<String, String> map = new HashMap<String, String>(); //使用jav的haspmap存储按照=切分后的key和value
for (int i = 0; i < x.length; i++) {
String[] temp = mapper.cutKeyValue(x); //按照=切分
if (temp.length == 1) {
map.put(temp[0], “”);
} else {
map.put(temp[0], temp[1]);
}
}
String hashkey = “pid”;
context.write(new Text(map.get(hashkey)), new IntWritable(1)); //输出 pid
}
　　reducer代码分析
　　public void reduce(Text pid, Iterable<IntWritable> values,
Context context) throws IOException, InterruptedException {
　　int pidCount = 0;
for (IntWritable value : values) {
pidCount++; //累加出现的次数
}
context.write(pid, new IntWritable(pidCount));
}
　　需要说明的是我们只记数，并不需要比较value，这点与上一个例子有所不同
4.按照之前介绍的方式打包，并执行以下命令
　　/root/hadoop-0.20.203.0/bin/hadoop jar /root/java_hadoop/PidCount.jar /root/log/ /root/out/
5.查看输出
　　cat /root/out/part-r-00000
50888   1
51224   1
51852   1
52088   1
52213   1
52249   2
52680   1
52929   1
53205   1
　　6.全代码展示
　　package hadoop.pidcount;
　　import java.io.IOException;
　　import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.io.BufferedInputStream;
import java.util.Scanner;
　　import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
　　public class NewPidCount {
　　static class NewPidCountMapper extends
Mapper<longwritable, text,="" intwritable=""> {
　　public void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
NewPidCountMapper mapper = new NewPidCountMapper();
String line = value.toString();
　　String[] x = mapper.cutLog(line.trim());
Map<string, string=""> map = new HashMap<string, string="">();
for (int i = 0; i < x.length; i++) {
String[] temp = mapper.cutKeyValue(x);
if (temp.length == 1) {
map.put(temp[0], "");
} else {
map.put(temp[0], temp[1]);
}
}
String hashkey = "pid";
context.write(new Text(map.get(hashkey)));
}
　　// 切分日志
public String[] cutLog(String string) {
String[] x = string.split("`");
return x;
}
　　// 切分key value
public String[] cutKeyValue(String string) {
String[] x = string.split("=");
return x;
}
}
　　static class NewPidCountReducer extends
Reducer<text, intwritable=""> {
　　public void reduce(Text pid,
Context context) throws IOException, InterruptedException {
　　Map<string, integer=""> pidCount = new HashMap<string, integer="">();
　　if (pidCount.containsKey(pid)) {
int count = pidCount.get(pid) + 1;
pidCount.remove(pid);
pidCount.put(pid, count);
} else {
pidCount.put(pid, 1);
}
context.write(pid, new IntWritable(pidCount.get(pid)));
}
}
　　public static void main(String[] args) throws Exception {
if (args.length != 2) {
System.err.println(“Usage: NewPidCount  “);
System.exit(-1);
}
　　Job job = new Job();
job.setJarByClass(NewPidCount.class);
　　FileInputFormat.addInputPath(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
　　job.setMapperClass(NewPidCountMapper.class);
job.setReducerClass(NewPidCountReducer.class);
　　job.setOutputKeyClass(Text.class);
job.setOutputValueClass(IntWritable.class);
　　System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}
　　转：http://yslion.com/?p=78