实验名称：Datajoin数据连接
实验目的：
1、记录我的Hadoop 实验过程，我是NCU HANG TIAN BAN 的学生。将会附上完整可运行的代码。程序中框架是一套模板百度的、书上也有但是重要算法是我自己写的将会标注。 http://blog.csdn.net/wawmg/article/details/8759076 这是我参考的框架模板。
2、提示大致浏览可看加粗部分【1、2、3、4】
实验要求：
任务1、多个数据源的内连接
【数据样例】
输入：
factory:
factoryname addressID
Beijing Red Star 1
Shenzhen Thunder 3
Guangzhou Honda 2
Beijing Rising 1
Guangzhou Development Bank 2
Tencent 3
Bank of Beijing 1
Nanchang Univ 5
address:
addressID addressname
1 Beijing
2 Guangzhou
3 Shenzhen
4 Xian
输出:
factorynameaddressIDaddressname
Bank of Beijing1Beijing
Beijing Red Star1Beijing
Beijing Rising1eijing
Guangzhou Development Bank2 Guangzhou
Guangzhou Honda2 Guangzhou
Shenzhen Thunder3 Shenzhen
Tencent3 Shenzhen
[代码开始了]【1】
// 先是TaggedWritable类 抄的不作改动
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import org.apache.hadoop.contrib.utils.join.TaggedMapOutput;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.util.ReflectionUtils;
/*TaggedMapOutput是一个抽象数据类型，封装了标签与记录内容
此处作为DataJoinMapperBase的输出值类型，需要实现Writable接口，所以要实现两个序列化方法
自定义输入类型*/
public class TaggedWritable extends TaggedMapOutput {
private Writable data;
public TaggedWritable() {
this.tag = new Text();
}
public TaggedWritable(Writable data) // 构造函数
{
this.tag = new Text(); // tag可以通过setTag()方法进行设置
this.data = data;
}
@Override
public void readFields(DataInput in) throws IOException {
tag.readFields(in);
String dataClz = in.readUTF();
if (this.data == null
|| !this.data.getClass().getName().equals(dataClz)) {
try {
this.data = (Writable) ReflectionUtils.newInstance(
Class.forName(dataClz), null);
} catch (ClassNotFoundException e) {
e.printStackTrace();
}
}
data.readFields(in);
}
@Override
public void write(DataOutput out) throws IOException {
tag.write(out);
out.writeUTF(this.data.getClass().getName());
data.write(out);
}
@Override
public Writable getData() {
return data;
}
}
// http://blog.csdn.net/wawmg/article/details/8759076
【2】Map阶段 算法自己写的
import org.apache.hadoop.contrib.utils.join.DataJoinMapperBase;
import org.apache.hadoop.contrib.utils.join.TaggedMapOutput;
import org.apache.hadoop.io.Text;
public class JoinMapper extends DataJoinMapperBase {
// 这个在任务开始时调用，用于产生标签
// 此处就直接以文件名作为标签
@Override
protected Text generateInputTag(String inputFile) {
System.out.println("inputFile = " + inputFile);
return new Text(inputFile);
}
// 这里我们已经确定分割符为‘,‘，更普遍的，用户应能自己指定分割符和组键。
// 设置组键
@Override
protected Text generateGroupKey(TaggedMapOutput record) {
String tag = ((Text) record.getTag()).toString();
if(tag.indexOf("factory") != -1){
String line = ((Text) record.getData()).toString();
String[] tokens = line.split(" ");
int len = tokens.length - 1;
return new Text(tokens[len]);
}else{
String line = ((Text) record.getData()).toString();
String[] tokens = line.split(" ");
return new Text(tokens[0]);
}
}
// 返回一个任何带任何我们想要的Text标签的TaggedWritable
@Override
protected TaggedMapOutput generateTaggedMapOutput(Object value) {
TaggedWritable retv = new TaggedWritable((Text) value);
retv.setTag(this.inputTag); // 不要忘记设定当前键值的标签
return retv;
}
}
【3】reduce阶段 算法也是自己写的
import org.apache.hadoop.contrib.utils.join.DataJoinReducerBase;
import org.apache.hadoop.contrib.utils.join.TaggedMapOutput;
import org.apache.hadoop.io.Text;
public class JoinReducer extends DataJoinReducerBase {
// 两个参数数组大小一定相同，并且最多等于数据源个数
@Override
protected TaggedMapOutput combine(Object[] tags, Object[] values) {
if (tags.length < 2) return null; // 这一步，实现内联结
String joinedStr = "";
String dd = " ";
for (int i = 0; i < values.length; i++) {
// 以逗号作为原两个数据源记录链接的分割符
TaggedWritable tw = (TaggedWritable) values[i];
String line = ((Text) tw.getData()).toString();
// 将一条记录划分两组，去掉第一组的组键名。
if( i == 0){
String[] tokens = line.split(" ");
dd += tokens[1];
}
if(i == 1){
joinedStr += line;
System.out.println(joinedStr);
}
}
joinedStr += dd;
TaggedWritable retv = new TaggedWritable(new Text(joinedStr));
retv.setTag((Text) tags[1]); // 这只retv的组键，作为最终输出键。
return retv;
}
}
【4】Driver 驱动类 抄的不作改动
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
public class DataJoinDriver extends Configured implements Tool {
public int run(String[] args) throws Exception {
Configuration conf = getConf();
Path in = new Path("hdfs://localhost:9000/user/c/input/*.txt");
Path out = new Path("hdfs://localhost:9000/user/c/output2");
JobConf job = new JobConf(conf, DataJoinDriver.class);
job.setJobName("DataJoin");
FileSystem hdfs = FileSystem.get(conf);
FileInputFormat.setInputPaths(job, in);
FileOutputFormat.setOutputPath(job, out);
job.setMapperClass(JoinMapper.class);
job.setReducerClass(JoinReducer.class);
job.setInputFormat(TextInputFormat.class);
job.setOutputFormat(TextOutputFormat.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(TaggedWritable.class);
JobClient.runJob(job);
return 0;
}
public static void main(String[] args) throws Exception {
int res = ToolRunner.run(new Configuration(), new DataJoinDriver(),
args);
System.exit(res);
}
}
最后：输出有点小问题，就是没有做排序。
原文地址：http://1660158374.blog.51cto.com/11537412/1782322
