Reduce Join
Map 端的主要工作:为来自不同表或文件的 key/value 对，打标签以区别不同来源的记 录。然后用连接字段作为 key，其余部分和新加的标志作为 value，最后进行输出；
Reduce 端的主要工作:在 Reduce 端以连接字段作为 key 的分组已经完成，我们只需要 在每一个分组当中将那些来源于不同文件的记录(在 Map 阶段已经打标志)分开，最后进 行合并就 ok 了。
Reduce Join案例
需求
订单表order：
产品表pd：
将上述两表中的数据根据商品pid合并到订单数据表中，要求呈现出如下图：
需求分析
通过将关联条件作为 Map 输出的 key，将两表满足 Join 条件的数据并携带数据所来源
的文件信息，发往同一个 ReduceTask，在 Reduce 中进行数据的串联。
Reduce端表合并(数据倾斜)
运行结果
本地order.txt
本地pd.txt
输出
缺点
缺点: 这种方式中，合并的操作是在 Reduce 阶段完成，Reduce 端的处理压力太大，Map节点的运算负载则很低，资源利用率不高，且在 Reduce 阶段极易产生数据倾斜。
解决方案: Map 端实现数据合并。
源码
tips:
hadoop迭代器中使用了对象重用，即迭代时value始终指向一个内存地址(引用值始终不变)，改变的是引用指向的内存地址中的数据。
TableBean类：
package com.xiaobai.mapreduce.reduceJoin;
import org.apache.hadoop.io.Writable;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
public class TableBean implements Writable {
private String id; //订单id
private String pid; //商品id
private int amount; //商品数量
private String pname; // 商品名称
private String flag; //标记表 order pd
//空餐构造
public TableBean() {
}
public String getId() {
return id;
}
public void setId(String id) {
this.id = id;
}
public String getPid() {
return pid;
}
public void setPid(String pid) {
this.pid = pid;
}
public int getAmount() {
return amount;
}
public void setAmount(int amount) {
this.amount = amount;
}
public String getPname() {
return pname;
}
public void setPname(String pname) {
this.pname = pname;
}
public String getFlag() {
return flag;
}
public void setFlag(String flag) {
this.flag = flag;
}
//序列化
@Override
public void write(DataOutput out) throws IOException {
out.writeUTF(id);
out.writeUTF(pid);
out.writeInt(amount);
out.writeUTF(pname);
out.writeUTF(flag);
}
//反序列化
@Override
public void readFields(DataInput in) throws IOException {
//反序列化应和序列化顺序一致
this.id = in.readUTF();
this.pid = in.readUTF();
this.amount = in.readInt();
this.pname = in.readUTF();
this.flag = in.readUTF();
}
@Override
public String toString() {
// id pname amount
return id + "\t" + pname + "\t" + amount;
}
}
TableMapper类：
package com.xiaobai.mapreduce.reduceJoin;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import java.io.File;
import java.io.IOException;
public class TableMapper extends Mapper<LongWritable, Text,Text,TableBean> {
private String fileName;
private Text outK = new Text();
private TableBean outV = new TableBean();
@Override
protected void setup(Context context) throws IOException, InterruptedException {
//初始化 order pd
//获取文件名称 一个文件只获取一次
FileSplit split = (FileSplit)context.getInputSplit();
fileName = split.getPath().getName();
}
@Override
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
//1. 获取一行
String line = value.toString();
//2.判断是哪个文件的
if(fileName.contains("order")){ //处理的是订单表order
String[] split = line.split("\t");
//封装
outK.set(split[1]);
outV.setId(split[0]);
outV.setPid(split[1]);
outV.setAmount(Integer.parseInt(split[2])); //要转换为String类型
outV.setPname("");
outV.setFlag("order");
}else{ //处理的是产品表pd
String[] split = line.split("\t");
outK.set(split[0]);
outV.setId("");
outV.setPid(split[0]);
outV.setAmount(0);
outV.setPname(split[1]);
outV.setFlag("pd");
}
//写出
context.write(outK,outV);
}
}
TableReducer类：
package com.xiaobai.mapreduce.reduceJoin;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
public class TableReducer extends Reducer<Text,TableBean,TableBean, NullWritable> {
@Override
protected void reduce(Text key, Iterable<TableBean> values, Context context) throws IOException, InterruptedException {
// 01 1001 1 order
// 01 1004 4 order
// 01 小米 pd
//创建2个集合
ArrayList<TableBean> orderBeans = new ArrayList<>();
TableBean pdBean = new TableBean();
//循环遍历
for (TableBean value : values) {
if("order".equals(value.getFlag())){ //order表
//创建临时TableBean对象tmptableBean
TableBean tmptableBean = new TableBean();
try {
BeanUtils.copyProperties(tmptableBean,value); //使用工具类BeanUtils将value赋值给tmptableBean
} catch (IllegalAccessException e) {
e.printStackTrace();
} catch (InvocationTargetException e) {
e.printStackTrace();
}
orderBeans.add(tmptableBean);
}else{ //pd表
try {
BeanUtils.copyProperties(pdBean,value);
} catch (IllegalAccessException e) {
e.printStackTrace();
} catch (InvocationTargetException e) {
e.printStackTrace();
}
}
}
//循环遍历orderBeans，赋值pdname
for (TableBean orderBean : orderBeans) {
orderBean.setPname(pdBean.getPname());//id相同
context.write(orderBean,NullWritable.get());
}
}
}
TableDriver类：
package com.xiaobai.mapreduce.reduceJoin;
import org.apache.hadoop.conf.Configuration;
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;
import org.apache.hadoop.fs.Path;
import java.io.IOException;
public class TableDriver {
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
Job job = Job.getInstance(new Configuration());
job.setJarByClass(TableDriver.class);
job.setMapperClass(TableMapper.class);
job.setReducerClass(TableReducer.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(TableBean.class);
job.setOutputKeyClass(TableBean.class);
job.setOutputValueClass(NullWritable.class);
FileInputFormat.setInputPaths(job,new org.apache.hadoop.fs.Path("/Users/jane/Desktop/test/JoinTest"));
FileOutputFormat.setOutputPath(job,new Path("/Users/jane/Desktop/hadoop/JoinTestOutput"));
boolean b = job.waitForCompletion(true);
System.exit(b ? 0 : 1);
}
}