场景：有两张表，一张用户表（user）,交易表（transactions）。两张表的字段如下：

两份表数据做个左连接，查询出（商品名，地址）这种格式。

 这样就是相当于交易表是左表，不管怎么样数据都要保留，然后从右边里面查出来弥补左表。

效果如下：

 

 

 思路：写两个map,把两个表的数据都读进来，在reduce端进行连接，然后按照格式要求写出去。

（1）map1：读取transaction文件，封装为：

protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, PairOfStrings, PairOfStrings>.Context context)

throws IOException, InterruptedException {

              String lines=value.toString();

              String[] args=lines.split(" ");

              String productID=args[1];

              String userID=args[2];

    //把outPutKey加了一个2，这么做的目的是，后续在reduce端，聚合时，这个数据能够晚点到达。

              outPutKey.set(userID, "2");

              outPutValue.set("P", productID);

              context.write(outPutKey, outPutValue);

}

 

 （2）map2：读取user文件，封装为：

static class map2 extends Mapper<LongWritable, Text,PairOfStrings,PairOfStrings>

{

PairOfStrings outPutKey=new PairOfStrings();

PairOfStrings outPutvalue=new PairOfStrings();

@Override

protected void map(LongWritable key, Text value,

Mapper<LongWritable, Text, PairOfStrings, PairOfStrings>.Context context)

throws IOException, InterruptedException {

String line=value.toString();

String[] args=line.split(" ");

String userID=args[0];

String locationID=args[1];

 //把outPutKey加了一个1，这么做的目的是，后续在reduce端，聚合时，这个数据能够早于transaction文件里面的数据到达。

outPutKey.set(userID, "1");

outPutvalue.set("L", locationID);

context.write(outPutKey, outPutvalue);

}

 

(3)reduce:把map端的数据要根据用户ID分区,相同的用户ID写入到同一个分区，进而写入到同一个Reduce分区，然后在Reduce中根据PairOfStrings这个类的自己的排序规则对数据排序。因为前面对key做了处理（加了1，2），所以是用户的地址这些信息先到达reduce。，然后根据不同的分组，把数据写出来。 

 

 

---

---

---

---

---

总的代码结构：

LeftCmain:

package com.guigu.left;import java.io.IOException;import java.util.Iterator;import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.fs.Path;import org.apache.hadoop.io.LongWritable;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.MultipleInputs;import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;import edu.umd.cloud9.io.pair.PairOfStrings;public class LeftCmain {        //读取transaction文件    static class map1 extends Mapper
     
          {            PairOfStrings outPutKey=new PairOfStrings();        PairOfStrings outPutValue=new PairOfStrings();        @Override        protected void map(LongWritable key, Text value, Mapper
      
       .Context context)                throws IOException, InterruptedException {              String lines=value.toString();              String[] args=lines.split(" ");              String productID=args[1];              String userID=args[2];              outPutKey.set(userID, "2");              outPutValue.set("P", productID);              context.write(outPutKey, outPutValue);        }            }        //读取user文件    static class map2 extends Mapper
       
            {        PairOfStrings outPutKey=new PairOfStrings();        PairOfStrings outPutvalue=new PairOfStrings();        @Override        protected void map(LongWritable key, Text value,                Mapper
        
         .Context context)                throws IOException, InterruptedException {            String line=value.toString();            String[] args=line.split(" ");            String userID=args[0];            String locationID=args[1];            outPutKey.set(userID, "1");            outPutvalue.set("L", locationID);            context.write(outPutKey, outPutvalue);        }            }            /**     * 这个的关键点在于,取出的数据：要求先取出地址的数据。     * @author Sxq     *     */    static class reduce1 extends Reducer
         
              {        Text produceID=new Text();        Text localID=new Text("undefine");                @Override        protected void reduce(PairOfStrings arg0, Iterable
          
            Iterator1, Reducer
           
            .Context context) throws IOException, InterruptedException { Iterator
            
              iterator=Iterator1.iterator(); //由于做了二次排序，可以保证先得到的是地址的数据。 if(iterator.hasNext()) { PairOfStrings fisrPair=iterator.next(); // System.out.println("firstPair="+fisrPair.toString()); //如果是地址的信息，那就把他直接放出来 if(fisrPair.getLeftElement().equals("L")) { localID.set(fisrPair.getRightElement()); } } while(iterator.hasNext()) { PairOfStrings pairOfStrings=iterator.next(); //System.out.println(pairOfStrings.toString()); produceID.set(pairOfStrings.getRightElement()); System.out.println("prdouct:"+produceID.toString()+"localId:"+localID.toString()); //System.out.println(); context.write(produceID, localID); } } } public static void main(String[] args) throws Exception { Configuration conf = new Configuration(); Job job = Job.getInstance(conf); job.setJarByClass(LeftCmain.class); job.setMapperClass(map1.class); job.setReducerClass(reduce1.class); job.setMapOutputKeyClass(PairOfStrings.class); job.setMapOutputValueClass(PairOfStrings.class); job.setOutputValueClass(Text.class); job.setOutputKeyClass(Text.class); job.setSortComparatorClass(PairOfStrings.Comparator.class); // 在Reduce端设置分组，使得同一个用户在同一个组，然后做拼接。 job.setGroupingComparatorClass(SecondarySortGroupComparator.class); // 设置分区 job.setPartitionerClass(SecondarySortParitioner.class); // job.setOutputFormatClass(SequenceFileOutputFormat.class); Path transactions=new Path("/Users/mac/Desktop/transactions.txt"); MultipleInputs.addInputPath(job,transactions,TextInputFormat.class,map1.class); MultipleInputs.addInputPath(job,new Path("/Users/mac/Desktop/user.txt"), TextInputFormat.class,map2.class); FileOutputFormat.setOutputPath(job, new Path("/Users/mac/Desktop/flowresort")); boolean result = job.waitForCompletion(true); System.exit(result ? 0 : 1); } }
            
           
          
         
        
       
      
     

 

 

SecondarySortGroupComparator:

package com.guigu.left;import org.apache.hadoop.io.DataInputBuffer;import org.apache.hadoop.io.WritableComparable;import org.apache.hadoop.io.WritableComparator;import com.book.test1.CompositeKey;import edu.umd.cloud9.io.pair.PairOfStrings;/** * 不同分区的组聚合时，可以按照我们要的顺序来排列 * @author Sxq *WritableComparator */public class SecondarySortGroupComparator extends WritableComparator {    public  SecondarySortGroupComparator() {         super(PairOfStrings.class,true);    }        @Override    public int compare(WritableComparable a, WritableComparable b) {                PairOfStrings v1=(PairOfStrings)a;        PairOfStrings v2=(PairOfStrings)b;        return v1.getLeftElement().compareTo(v2.getLeftElement());    }}

SecondarySortParitioner:

package com.guigu.left;import org.apache.hadoop.mapreduce.Partitioner;import edu.umd.cloud9.io.pair.PairOfStrings;/** *  * @author Sxq * */public class SecondarySortParitioner extends Partitioner
     
      {    @Override    public int getPartition(PairOfStrings key, Object value, int numPartitions) {        return (key.getLeftElement().hashCode()&Integer.MAX_VALUE)%numPartitions;    }}
     

 

 

运行结果：