TableBean
public class TableBean implements Writable {
private String id; //订单id
private String pid; //产品id
private int amount; //数量
private String pname; //产品名称
private String flag; //定义一个标记，标记是产品表还是订单表
public TableBean() {
super();
}
public TableBean(String id, String pid, int amount, String pname, String flag) {
super();
this.id = id;
this.pid = pid;
this.amount = amount;
this.pname = pname;
this.flag = flag;
}
public String getId() {
return id;
}
public void setId(String id) {
this.id = id;
}
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 getPid() {
return pid;
}
public void setPid(String pid) {
this.pid = pid;
}
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 {
//反序列化方法
id = in.readUTF();
pid = in.readUTF();
amount = in.readInt();
pname = in.readUTF();
flag = in.readUTF();
}
@Override
public String toString() {
return id + '\t' + amount + "\t" + pname;
}
}
TableMapper
public class TableMapper extends Mapper<LongWritable, Text, Text, TableBean> {
String name;
TableBean tableBean = new TableBean();
Text k = new Text();
@Override
protected void setup(Context context) throws IOException, InterruptedException {
//获取文件的名称
FileSplit inputSplit = (FileSplit) context.getInputSplit();
name = inputSplit.getPath().getName();
}
@Override
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
// 1 获取一行
String line = value.toString();
if (name.startsWith("order")) { //订单表
String[] fields = line.split("\t");
//封装key和value
tableBean.setId(fields[0]);
tableBean.setPid(fields[1]);
tableBean.setAmount(Integer.parseInt(fields[2]));
tableBean.setPname("");
tableBean.setFlag("order");
k.set(fields[1]);
} else { //产品表
String[] fields = line.split("\t");
//封装key和value
tableBean.setId("");
tableBean.setPid(fields[0]);
tableBean.setAmount(0);
tableBean.setPname(fields[1]);
tableBean.setFlag("pd");
k.set(fields[0]);
}
//写出
context.write(k, tableBean);
}
}
TableReducer
public class TableReducer extends Reducer<Text, TableBean, TableBean, NullWritable> {
@Override
protected void reduce(Text key, Iterable<TableBean> values, Context context) throws IOException, InterruptedException {
//存储所有订单集合
ArrayList<TableBean> orderBeans = new ArrayList<>();
//存储产品信息
TableBean pdBean = new TableBean();
for (TableBean tableBean : values) {
if ("order".equals(tableBean.getFlag())) { //订单表
TableBean tmpBean = new TableBean();
try {
BeanUtils.copyProperties(tmpBean, tableBean);
orderBeans.add(tmpBean);
} catch (IllegalAccessException e) {
e.printStackTrace();
} catch (InvocationTargetException e) {
e.printStackTrace();
}
} else {
try {
BeanUtils.copyProperties(pdBean, tableBean);
} catch (IllegalAccessException e) {
e.printStackTrace();
} catch (InvocationTargetException e) {
e.printStackTrace();
}
}
}
for (TableBean tableBean:orderBeans) {
tableBean.setPname(pdBean.getPname());
context.write(tableBean, NullWritable.get());
}
}
}
TableDriver
public class TableDriver {
public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
//输入输出路径需要根据自己电脑上的实际的输入输出路径设置
args = new String[]{"e:/input/inputtable", "e:/output1"};
// 1 获取配置信息
Configuration conf = new Configuration();
Job job = Job.getInstance(conf);
// 2 设置jar包加载路径
job.setJarByClass(TableDriver.class);
// 3 加载map/reduce类
job.setMapperClass(TableMapper.class);
job.setReducerClass(TableReducer.class);
// 4 设置map输出数据kv类型
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(TableBean.class);
// 5 设置最终输出数据的kv类型
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(NullWritable.class);
// 6 设置输入数据和输出数据路径
FileInputFormat.setInputPaths(job, new Path(args[0]));
FileOutputFormat.setOutputPath(job, new Path(args[1]));
// 8 设置reduce端的分组
job.setGroupingComparatorClass(OrderGroupingComparator.class);
// 7 提交job
boolean result = job.waitForCompletion(true);
System.exit(result ? 0 : 1);
}
}
网友评论