Mapreduce实例——Reduce端join
在Reudce端进行Join连接是MapReduce框架进行表之间Join操作最为常见的模式。
1.Reduce端Join实现原理
（1）Map端的主要工作，为来自不同表（文件）的key/value对打标签以区别不同来源的记录。然后用连接字段作为key，其余部分和新加的标志作为value，最后进行输出。
（2）Reduce端的主要工作，在Reduce端以连接字段作为key的分组已经完成，我们只需要在每一个分组当中将那些来源于不同文件的记录（在map阶段已经打标志）分开，最后进行笛卡尔只就ok了。
2.Reduce端Join的使用场景
Reduce端连接比Map端连接更为普遍，因为在map阶段不能获取所有需要的join字段，即：同一个key对应的字段可能位于不同map中，但是Reduce端连接效率比较低，因为所有数据都必须经过Shuffle过程。
3.本实验的Reduce端Join代码执行流程：
（1）Map端读取所有的文件，并在输出的内容里加上标识，代表数据是从哪个文件里来的。
（2）在Reduce处理函数中，按照标识对数据进行处理。
（3）然后将相同的key值进行Join连接操作，求出结果并直接输出。
(1)Map端读取所有的文件，并在输出的内容里加上标识，代表数据是从哪个文件里来的。
(2)在reduce处理函数中，按照标识对数据进行处理。
(3)然后将相同key值进行join连接操作，求出结果并直接输出。
Mapreduce中join连接分为Map端Join与Reduce端Join，这里是一个Reduce端Join连接。程序主要包括两部分：Map部分和Reduce部分。
Map处理的是一个纯文本文件，Mapper处理的数据是由InputFormat将数据集切分成小的数据集InputSplit，并用RecordReader解析成<key,value>对提供给map函数使用。在map函数中，首先用getPath()方法获取分片InputSplit的路径并赋值给filePath，if判断filePath中如果包含goods.txt文件名，则将map函数输入的value值通过Split("\t")方法进行切分，与goods_visit文件里相同的商品id字段作为key，其他字段前加"1+"作为value。如果if判断filePath包含goods_visit.txt文件名，步骤与上面相同，只是把其他字段前加"2+"作为value。最后把<key,value>通过Context的write方法输出。
map函数输出的<key,value>经过shuffle将key相同的所有value放到一个迭代器中形成values，然后将<key,values>键值对传递给reduce函数。reduce函数中，首先新建两个Vector集合，用于存放输入的values中以"1+"开头和"2+"开头的数据。然后用增强版for循环遍历并嵌套if判断，若判断values里的元素以1+开头，则通过substring(2)方法切分元素，结果存放到left集合中，若values里元素以2+开头，则仍利用substring(2)方法切分元素，结果存放到right集合中。最后再用两个嵌套for循环，遍历输出<key,value>，其中输入的key直接赋值给输出的key，输出的value为left +"\t"+right。
代码如下:
package exper;
import java.io.IOException;
import java.util.Iterator;
import java.util.Vector;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
public class ReduceJoin {
public static class mymapper extends Mapper<Object, Text, Text, Text> {
@Override
protected void map(Object key, Text value, Context context)
throws IOException, InterruptedException {
String filePath = ((FileSplit)
context.getInputSplit()).getPath().toString();
if (filePath.contains("orders1")) {
String line = value.toString();
String[] arr = line.split(" ");
context.write(new Text(arr[0]), new Text("1+" + arr[2] + "\t" + arr[3]));
//System.out.println(arr[0] +
"_1+" + arr[2]+"\t"+arr[3]);
} else if (filePath.contains("order_items1")) {
String line = value.toString();
String[] arr = line.split(" ");
context.write(new Text(arr[1]), new Text("2+" + arr[2]));
//System.out.println(arr[1] +
"_2+" + arr[2]);
}
}
}
public static class myreducer extends Reducer<Text, Text, Text, Text> {
@Override
protected void reduce(Text key, Iterable<Text> values, Context context)
throws IOException, InterruptedException {
Vector<String> left = new Vector<String>();
Vector<String> right = new Vector<String>();
for (Text val : values) {
String str = val.toString();
if (str.startsWith("1+")) {
left.add(str.substring(2));
} else if (str.startsWith("2+")) {
right.add(str.substring(2));
}
}
int sizeL = left.size();
int sizeR = right.size();
//System.out.println(key +
"left:"+left);
//System.out.println(key +
"right:"+right);
for (int i = 0; i < sizeL; i++) {
for (int j = 0; j < sizeR; j++) {
context.write(key, new Text(left.get(i) + "\t" + right.get(j)));
//System.out.println(key +
" \t" +
left.get(i) + "\t" +
right.get(j));
}
}
}
}
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
Job job = Job.getInstance();
job.setJobName("reducejoin");
job.setJarByClass(ReduceJoin.class);
job.setMapperClass(mymapper.class);
job.setReducerClass(myreducer.class);
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(Text.class);
Path left = new Path("D:\\mapreduce\\5in\\order1");
Path right = new Path("D:\\mapreduce\\5in\\order_item1");
Path out = new Path("file:///D:/mapreduce/6out");
FileInputFormat.addInputPath(job,left);
FileInputFormat.addInputPath(job, right);
FileOutputFormat.setOutputPath(job,out);
System.exit(job.waitForCompletion(true) ? 0 : 1);
}
}