HADOOP JAVA MapReduce API

下面结合具体业务讲解一下 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[i]); //按照=切分
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[i]);
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