Hadoop 实例9 Join讲解2: 将人员的地址ID完善成为地址名称

输出格式要求：人员Id，姓名，地址

1、原始数据

人员ID 人员名称 地址ID

1   张三  1
2   李四  2
3   王五  1
4   赵六  3
5   马七  3

另外一组为地址信息:

地址ID 地址名称

1   北京
2   上海
3   广州

2、处理说明

这里给出了一个很简单的例子,而且数据量很小,就这么用眼睛就能看过来的几行,当然,实际的情况可能是几十万上百万甚至上亿的数据量.

要实现的功能很简单, 就是将人员信息与地址信息进行join,将人员的地址ID完善成为地址名称.

对于Hadoop文件系统的应用,目前看来,很多数据的存储都是基于文本的, 而且都是将数据放在一个文件目录中进行处理.因此我们这里也采用这种模式来完成.

对于mapreduce程序来说,最主要的就是将要做的工作转化为map以及reduce两个部分.

我们可以将地址以及人员都采用同样的数据结构来存储,通 过一个flag标志来指定该数据结构里面存储的是地址信息还是人员信息.

经过map后,使用地址ID作为key,将所有的具有相同地址的地址信息和人员信 息放入一个key->value list数据结构中传送到reduce中进行处理.

在reduce过程中,由于key是地址的ID,所以value list中只有一个是地址信息,其他的都是人员信息,因此,找到该地址信息后,其他的人员信息的地址就是该地址所指定的地址名称.

OK,我们的join算法基本搞定啦.剩下就是编程实现了,let’s go.

3、中间bean实现

上面提到了存储人员和地址信息的数据结构,可以说这个数据结构是改程序的重要的数据载体之一.我们先来看看该数据结构:

package cn.edu.bjut.jointwo;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;

import org.apache.hadoop.io.WritableComparable;

public class Member implements WritableComparable<Member>{

private String userNo = "";
private String userName = "";
private String cityNo = "";
private String cityName = "";
private int flag = 0;

public Member() {
super();
}

public Member(String userNo, String userName, String cityNo,
String cityName, int flag) {
super();
this.userNo = userNo;
this.userName = userName;
this.cityNo = cityNo;
this.cityName = cityName;
this.flag = flag;
}

public Member(Member m) {
super();
this.userNo = m.getUserNo();
this.userName = m.getUserName();
this.cityNo = m.getCityNo();
this.cityName = m.getCityName();
this.flag = m.getFlag();
}

public void write(DataOutput out) throws IOException {
out.writeUTF(getUserNo());
out.writeUTF(getUserName());
out.writeUTF(getCityNo());
out.writeUTF(getCityName());
out.writeInt(getFlag());
}

public void readFields(DataInput in) throws IOException {
this.userNo = in.readUTF();
this.userName = in.readUTF();
this.cityNo = in.readUTF();
this.cityName = in.readUTF();
this.flag = in.readInt();
}

public int compareTo(Member o) {
return 0;
}

@Override
public String toString() {
return "userNo=" + userNo + ", userName=" + userName
+ ", cityNo=" + cityNo + ", cityName=" + cityName;
}

public String getUserNo() {
return userNo;
}

public void setUserNo(String userNo) {
this.userNo = userNo;
}

public String getUserName() {
return userName;
}

public void setUserName(String userName) {
this.userName = userName;
}

public String getCityNo() {
return cityNo;
}

public void setCityNo(String cityNo) {
this.cityNo = cityNo;
}

public String getCityName() {
return cityName;
}

public void setCityName(String cityName) {
this.cityName = cityName;
}

public int getFlag() {
return flag;
}

public void setFlag(int flag) {
this.flag = flag;
}

}

4.Mapper程序：

package cn.edu.bjut.jointwo;

import java.io.IOException;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class JoinMapper extends Mapper<LongWritable, Text, Text, Member> {

@Override
protected void map(LongWritable key, Text value, Context context)
throws IOException, InterruptedException {
String line = value.toString();
String[] arr = line.split("\t");
if(arr.length >= 3) {
Member m = new Member();
m.setUserNo(arr[0]);
m.setUserName(arr[1]);
m.setCityNo(arr[2]);
m.setFlag(0);

context.write(new Text(m.getCityNo()), m);
} else {
Member m = new Member();
m.setCityNo(arr[0]);
m.setCityName(arr[1]);
m.setFlag(1);

context.write(new Text(m.getCityNo()), m);
}
}

}

5.Reducer程序：

package cn.edu.bjut.jointwo;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class JoinReducer extends Reducer<Text, Member, Text, NullWritable> {

@Override
protected void reduce(Text key, Iterable<Member> values, Context context)
throws IOException, InterruptedException {
Member m = null;
List<Member> list = new ArrayList<Member>();

for(Member member : values) {
if(0 == member.getFlag()) {
list.add(new Member(member));
} else {
m = new Member(member);
}
}

if(null != m) {
for(Member member : list) {
member.setCityName(m.getCityName());
context.write(new Text(member.toString()), NullWritable.get());
}
}
}

}

6.主程序：

package cn.edu.bjut.jointwo;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class MainJob {

public static void main(String[] args) throws Exception {
Configuration conf = new Configuration();
Job job = new Job(conf, "jointwo");
job.setJarByClass(MainJob.class);

job.setMapperClass(JoinMapper.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(Member.class);

job.setReducerClass(JoinReducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(NullWritable.class);

FileInputFormat.addInputPath(job, new Path(args[0]));

Path outPathDir = new Path(args[1]);
FileSystem fs = FileSystem.get(conf);
if(fs.exists(outPathDir)) {
fs.delete(outPathDir, true);
}

FileOutputFormat.setOutputPath(job, outPathDir);

job.waitForCompletion(true);
}

}