DataJoin类 实现不同格式数据reduce侧连接

实验名称：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);
}

}

最后：输出有点小问题，就是没有做排序。