需求
JavaBean类
package Join.reduceJoin.Demo2;
import org.apache.hadoop.io.Writable;
import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
public class JavaBean implements Writable {
private String id;
private String date;
private String pid;
private String amount;
private String name;
private String categoryId;
private String price;
public JavaBean() {
}
public JavaBean(String name, String categoryId, String price) {
this.name = name;
this.categoryId = categoryId;
this.price = price;
}
public JavaBean(String id, String date, String pid, String amount) {
this.id = id;
this.date = date;
this.pid = pid;
this.amount = amount;
}
@Override
public String toString() {
return id+","+date+","+pid+","+amount+","+name+","+categoryId+","+price;
}
public String getId() {
return id;
}
public void setId(String id) {
this.id = id;
}
public String getDate() {
return date;
}
public void setDate(String date) {
this.date = date;
}
public String getPid() {
return pid;
}
public void setPid(String pid) {
this.pid = pid;
}
public String getAmount() {
return amount;
}
public void setAmount(String amount) {
this.amount = amount;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public String getCategoryId() {
return categoryId;
}
public void setCategoryId(String categoryId) {
this.categoryId = categoryId;
}
public String getPrice() {
return price;
}
public void setPrice(String price) {
this.price = price;
}
@Override
public void write(DataOutput out) throws IOException {
out.writeUTF(id+"");
out.writeUTF(date+"");
out.writeUTF(pid+"");
out.writeUTF(amount+"");
out.writeUTF(name+"");
out.writeUTF(categoryId+"");
out.writeUTF(price+"");
}
@Override
public void readFields(DataInput in) throws IOException {
this.id = in.readUTF();
this.date = in.readUTF();
this.pid = in.readUTF();
this.amount = in.readUTF();
this.name = in.readUTF();
this.categoryId = in.readUTF();
this.price = in.readUTF();
}
}
Map类
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import java.io.IOException;
public class Map extends Mapper<LongWritable, Text,Text,JavaBean> {
@Override
protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
FileSplit fileSplit = (FileSplit) context.getInputSplit();
String name = fileSplit.getPath().getName();
if (name.contains("orders")){
String[] arr1 = value.toString().split(",");
context.write(new Text(arr1[2]),new JavaBean(arr1[0],arr1[1],arr1[2],arr1[3]));
}else if (name.contains("product")){
String[] arr2 = value.toString().split(",");
context.write(new Text(arr2[0]),new JavaBean(arr2[1],arr2[2],arr2[3]));
}
}
}
Reduce类
package Join.reduceJoin.Demo2;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;
import java.io.IOException;
public class Reduce extends Reducer<Text,JavaBean,JavaBean, NullWritable> {
@Override
protected void reduce(Text key, Iterable<JavaBean> values, Context context) throws IOException, InterruptedException {
JavaBean jb = new JavaBean();
for (JavaBean value : values) {
System.out.println(value.toString());
if (value.getId()!=null ){
jb.setPid(value.getPid());
jb.setId(value.getId());
jb.setDate(value.getDate());
jb.setAmount(value.getAmount());
}else {
jb.setName(value.getName());
jb.setCategoryId(value.getCategoryId());
jb.setPrice(value.getPrice());
}
}
context.write(jb,NullWritable.get());
}
}
Drive类
package Join.reduceJoin.Demo2;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
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;
public class Drive {
public static void main(String[] args) throws Exception{
Configuration conf = new Configuration();
Job job = new Job(conf, "joinDemo2");
job.setJarByClass(Drive.class);
job.setMapperClass(Map.class);
job.setReducerClass(Reduce.class);
job.setMapOutputKeyClass(Text.class);
job.setMapOutputValueClass(JavaBean.class);
job.setOutputKeyClass(JavaBean.class);
job.setOutputValueClass(NullWritable.class);
FileInputFormat.addInputPath(job,new Path("E:\\传智\\大二上半年\\10月\\预习资料\\05MapReduce\\素材\\素材\\4\\map端join\\input"));
FileOutputFormat.setOutputPath(job,new Path("F:\\桌面\\cc"));
boolean b = job.waitForCompletion(true);
System.exit(b?0:1);
}
}