背景：现在有两张表：customer、order，他们有个共同的cid,需要通过cid实现两张表的连接，并且通过cid进行分组排序

思路：首先通过mapper从context里面获取到文件切片，从文件切片中得到路径，从而判断是customer.txt,还是order.txt

然后依次读取每行内容，并且为每行数据打上一个标签，0表示customer，1表示order，最后组成一个新的组合key

然后调用排序对比器的方法，按照cid进行升序排列，同一个客户的order进行升序排列

然后进行reducer，将客户信息和订单信息进行拼接，最后进行cid分组，同一个客户的信息为一组，并按照cid升序排列

最后根据cid进行分区

comkey

package com.cr.reduceJoin;import org.apache.hadoop.io.WritableComparable;import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;public class comkey_1 implements WritableComparable<comkey_1>{//0表示客户信息 1表示订单信息private int type;private int cid;private int oid;private String customerInfo = "";private String orderInfo = "";public int getType() {return type;}public void setType(int type) {this.type = type;}public int getCid() {return cid;}public void setCid(int cid) {this.cid = cid;}public int getOid() {return oid;}public void setOid(int oid) {this.oid = oid;}public String getCustomerInfo() {return customerInfo;}public void setCustomerInfo(String customerInfo) {this.customerInfo = customerInfo;}public String getOrderInfo() {return orderInfo;}public void setOrderInfo(String orderInfo) {this.orderInfo = orderInfo;}@Overridepublic int compareTo(comkey_1 o) {int type1 = o.type;int cid1 = o.cid;int oid1 = o.oid;String customerInfo1 = o.customerInfo;String orderInfo1 = o.orderInfo;//首先判断是否是同一个客户的数据if(cid1 == cid){//判断是否是同一个客户的两个订单if(type1 == type){//升序排列return oid - oid1;}else {//如果一个是客户一个是订单，让客户信息在前if(type == 0){return -1;}elsereturn  1;}}else {//如果是不同客户，升序排列return cid - cid1;}}@Overridepublic void write(DataOutput out) throws IOException {out.writeInt(type);out.writeInt(cid);out.writeInt(oid);out.writeUTF(customerInfo);out.writeUTF(orderInfo);}@Overridepublic void readFields(DataInput in) throws IOException {this.type = in.readInt();this.cid = in.readInt();this.oid = in.readInt();this.customerInfo = in.readUTF();this.orderInfo = in.readUTF();}
}

mapper

package com.cr.reduceJoin;import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
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.IOException;public class ReduceJoinMapper extends Mapper<LongWritable,Text,comkey_1,NullWritable> {@Overrideprotected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {String line = value.toString();//从环境变量中获取文件切片，从文件切片中获取路径InputSplit inputSplit = context.getInputSplit();FileSplit fileSplit = (FileSplit)inputSplit;String path = fileSplit.getPath().toString();comkey_1 comkey1 = new comkey_1();//整理文本信息成为自己定义的comkey的格式if(path.contains("customer")){String cid = line.substring(0, line.indexOf(","));String cusInfo = line;comkey1.setType(0);comkey1.setCid(Integer.parseInt(cid));comkey1.setCustomerInfo(cusInfo);}else {String cid = line.substring(line.lastIndexOf(",") + 1);String oInfo = line.substring(0, line.lastIndexOf(","));String oid = line.substring(0, line.indexOf(","));comkey1.setType(1);comkey1.setCid(Integer.parseInt(cid));comkey1.setOid(Integer.parseInt(oid));comkey1.setOrderInfo(oInfo);}//写入上下文context.write(comkey1,NullWritable.get());}
}

组合key排序对比器

package com.cr.reduceJoin;import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;/*** 组合key排序对比器*/
public class Comkey2Comparator extends WritableComparator {protected Comkey2Comparator(){super(comkey_1.class,true);}@Overridepublic int compare(WritableComparable a, WritableComparable b) {comkey_1 a1 = (comkey_1) a;comkey_1 b1 = (comkey_1) b;return a1.compareTo(b1);}
}

reducer

package com.cr.reduceJoin;import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;import java.io.IOException;
import java.util.Iterator;public class JoinReducer  extends Reducer<comkey_1,NullWritable,Text,NullWritable>{@Overrideprotected void reduce(comkey_1 key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {//获取迭代器Iterator<NullWritable> iterator = values.iterator();//第一条一定是客户信息iterator.next();int type = key.getType();int cid = key.getCid();String customerInfo = key.getCustomerInfo();//从第二条开始就是订单信息while (iterator.hasNext()){iterator.next();String orderInfo = key.getOrderInfo();//按照客户+订单的方式进行拼接context.write(new Text(customerInfo + "," + orderInfo),NullWritable.get());}}
}

cid分组对比器

package com.cr.reduceJoin;import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.WritableComparator;/*
CID分组对比器*/
public class CIDgroupComparator extends WritableComparator {protected CIDgroupComparator(){super(comkey_1.class,true);}@Overridepublic int compare(WritableComparable a, WritableComparable b) {comkey_1 a1 = (comkey_1) a;comkey_1 b1 = (comkey_1) b;return a1.getCid() - b1.getCid();}
}

根据cid分区

package com.cr.reduceJoin;import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Partitioner;public class CIDpartition extends Partitioner<comkey_1,NullWritable>{//自定义分区类，按照cid分区@Overridepublic int getPartition(comkey_1 comkey_1, NullWritable nullWritable, int i) {return comkey_1.getCid() % i ;}
}

app

package com.cr.reduceJoin;import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
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;import java.io.IOException;public class ReduceJoinApp {public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {Configuration conf = new Configuration();conf.set("fs.defaultFS","file:///");Job job = Job.getInstance(conf);job.setJobName("ReduceJoinApp");job.setJarByClass(ReduceJoinApp.class);FileInputFormat.addInputPath(job,new Path(args[0]));Path out = new Path(args[1]);FileSystem fs = FileSystem.get(conf);if (fs.exists(out)) {fs.delete(out, true);}FileOutputFormat.setOutputPath(job,new Path(args[1]));job.setMapperClass(ReduceJoinMapper.class);job.setReducerClass(JoinReducer.class);job.setMapOutputKeyClass(comkey_1.class);job.setMapOutputValueClass(NullWritable.class);job.setOutputKeyClass(Text.class);job.setOutputValueClass(NullWritable.class);job.setPartitionerClass(CIDpartition.class);job.setGroupingComparatorClass(CIDgroupComparator.class);job.setSortComparatorClass(Comkey2Comparator.class);job.setNumReduceTasks(2);job.waitForCompletion(true);}
}

part-r-000                                   part-r-001