问题描述：

一个trade table表

product1"trade1

product2"trade2

product3"trade3

一个pay table表

product1"pay1

product2"pay2

product2"pay3

product1"pay4

product3"pay5

product3"pay6

建立两个表之间的连接，该两表是一对多关系的

如下：

trade1pay1

trade1pay4

trade2pay2

...

思路：

       为了将两个表整合到一起，由于有相同的第一列，且第一个表与第二个表是一对多关系的。

这里依然采用分组，以及组内排序，只要保证一方最先到达reduce端，则就可以进行迭代处理了。

为了保证第一个表先到达reduce端，可以为定义一个组合键，包含两个值，第一个值为product，第二个值为0或者1，来分别代表第一个表和第二个表，只要按照组内升序排列即可。

具体代码：

自定义组合键策略

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
package  whut.onetomany;
import  java.io.DataInput;
import  java.io.DataOutput;
import  java.io.IOException;
import  org.apache.hadoop.io.WritableComparable;
public  class  TextIntPair  implements  WritableComparable{
     //product1 0/1
     private  String firstKey; //product1
     private  int  secondKey; //0,1;0代表是trade表，1代表是pay表
     //只需要保证trade表在pay表前面就行，则只需要对组顺序排列
                                                           
     public  String getFirstKey() {
         return  firstKey;
     }
     public  void  setFirstKey(String firstKey) {
         this .firstKey = firstKey;
     }
     public  int  getSecondKey() {
         return  secondKey;
     }
     public  void  setSecondKey( int  secondKey) {
         this .secondKey = secondKey;
     }
     @Override
     public  void  write(DataOutput out)  throws  IOException {
         out.writeUTF(firstKey);
         out.writeInt(secondKey);
     }
     @Override
     public  void  readFields(DataInput in)  throws  IOException {
         // TODO Auto-generated method stub
         firstKey=in.readUTF();
         secondKey=in.readInt();
     }
                                                           
     @Override
     public  int  compareTo(Object o) {
         // TODO Auto-generated method stub
         TextIntPair tip=(TextIntPair)o;
         return  this .getFirstKey().compareTo(tip.getFirstKey());
     }
}

分组策略

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
package  whut.onetomany;
import  org.apache.hadoop.io.WritableComparable;
import  org.apache.hadoop.io.WritableComparator;
public  class  TextComparator  extends  WritableComparator{
     protected  TextComparator() {
         super (TextIntPair. class , true ); //注册比较器
     }
     @Override
     public  int  compare(WritableComparable a, WritableComparable b) {
         // TODO Auto-generated method stub
         TextIntPair tip1=(TextIntPair)a;
         TextIntPair tip2=(TextIntPair)b;
         return  tip1.getFirstKey().compareTo(tip2.getFirstKey());
     }
}

组内排序策略：目的是保证第一个表比第二个表先到达

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
package  whut.onetomany;
import  org.apache.hadoop.io.WritableComparable;
import  org.apache.hadoop.io.WritableComparator;
//分组内部进行排序，按照第二个字段进行排序
public  class  TextIntComparator  extends  WritableComparator {
     public  TextIntComparator()
     {
         super (TextIntPair. class , true );
     }
     //这里可以进行排序的方式管理
     //必须保证是同一个分组的
     //a与b进行比较
     //如果a在前b在后，则会产生升序
     //如果a在后b在前，则会产生降序
     @Override
     public  int  compare(WritableComparable a, WritableComparable b) {
         // TODO Auto-generated method stub
         TextIntPair ti1=(TextIntPair)a;
         TextIntPair ti2=(TextIntPair)b;
         //首先要保证是同一个组内，同一个组的标识就是第一个字段相同
         if (!ti1.getFirstKey().equals(ti2.getFirstKey()))
            return  ti1.getFirstKey().compareTo(ti2.getFirstKey());
         else
            return  ti1.getSecondKey()-ti2.getSecondKey(); //0,-1,1
     }
                                      
}

分区策略：

1
2
3
4
5
6
7
8
9
10
package  whut.onetomany;
import  org.apache.hadoop.io.Text;
import  org.apache.hadoop.mapreduce.Partitioner;
public  class  PartitionByText  extends  Partitioner<TextIntPair, Text> {
     @Override
     public  int  getPartition(TextIntPair key, Text value,  int  numPartitions) {
         // TODO Auto-generated method stub
         return  (key.getFirstKey().hashCode()&Integer.MAX_VALUE)%numPartitions;
     }
}

MapReduce

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
package  whut.onetomany;
import  java.io.IOException;
import  java.util.Iterator;
import  org.apache.hadoop.conf.Configuration;
import  org.apache.hadoop.conf.Configured;
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.Mapper.Context;
import  org.apache.hadoop.mapreduce.Reducer;
import  org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import  org.apache.hadoop.mapreduce.lib.input.FileSplit;
import  org.apache.hadoop.mapreduce.lib.input.MultipleInputs;
import  org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import  org.apache.hadoop.util.GenericOptionsParser;
import  org.apache.hadoop.util.Tool;
import  org.apache.hadoop.util.ToolRunner;
public  class  JoinMain  extends  Configured  implements  Tool {
     public  static  class  JoinMapper  extends  Mapper<LongWritable, Text, TextIntPair, Text>
     {
         private  TextIntPair tp= new  TextIntPair();
         private  Text val= new  Text();
         @Override
         protected  void  map(LongWritable key, Text value, Context context)
                 throws  IOException, InterruptedException {
             // TODO Auto-generated method stub
             //获取要处理的文件的名称
             FileSplit file=(FileSplit)context.getInputSplit();
             String fileName=file.getPath().toString();
             //获取输入行分隔
             String line=value.toString();
             String[] lineKeyValue=line.split( "\"" );
             String lineKey=lineKeyValue[ 0 ];
             String lineValue=lineKeyValue[ 1 ];
             tp.setFirstKey(lineKey);
             //判断是否是trade文件
             if (fileName.indexOf( "trade" )>= 0 )
             {
                 tp.setSecondKey( 0 );
                 val.set(lineValue);
             }
             //判断是否是pay文件
             else  if (fileName.indexOf( "pay" )>= 0 )
             {
                 tp.setSecondKey( 1 );
                 val.set(lineValue);
             }
             context.write(tp, val);
         }
     }
                       
     public  static  class  JoinReducer  extends  Reducer<TextIntPair, Text, Text, Text>
     {
         @Override
         protected  void  reduce(TextIntPair key, Iterable<Text> values,
                 Context context) throws  IOException, InterruptedException {
             Iterator<Text> valList=values.iterator();
             //注意这里一定要写成string不可变，写成Text有问题
             //Text trade=valList.next();
             String tradeName=valList.next().toString();
             while (valList.hasNext())
             {
                 Text pay=valList.next();
                 context.write( new  Text(tradeName), pay);
             }
         }
     }
     @Override
     public  int  run(String[] args)  throws  Exception
     {
         Configuration conf=getConf();
         Job job= new  Job(conf, "JoinJob" );
         job.setJarByClass(JoinMain. class );
         //ToolRunner已经利用GenericOptionsParser解析了命令行中的参数
         //并且将其存放在数组中，传递给该run()方法了
         FileInputFormat.addInputPath(job,  new  Path(args[ 0 ]));
         FileInputFormat.addInputPath(job,  new  Path(args[ 1 ]));
         //输入文件必须以，隔开
         //FileInputFormat.addInputPaths(job, args[0]);
         FileOutputFormat.setOutputPath(job,  new  Path(args[ 2 ]));
                           
         job.setMapperClass(JoinMapper. class );
         job.setReducerClass(JoinReducer. class );
         //设置分区方法
         job.setPartitionerClass(PartitionByText. class );
         //设置分组排序
         job.setGroupingComparatorClass(TextComparator. class );
         job.setSortComparatorClass(TextIntComparator. class );
                           
         job.setMapOutputKeyClass(TextIntPair. class );
         job.setMapOutputValueClass(Text. class );
         job.setOutputKeyClass(Text. class );
         job.setOutputValueClass(Text. class );
         job.waitForCompletion( true );
         int  exitCode=job.isSuccessful()? 0 : 1 ;
         return  exitCode;
     }
     public  static  void  main(String[] args) throws  Exception
     {
         // TODO Auto-generated method stub
         int  code=ToolRunner.run( new  JoinMain(), args);
         System.exit(code);
     }
}


注意：

     一般有些地方没有定义组内排序策略，但是经过多次测试，发现无法保证第一个表在第二个表之前到达，则这里就自定义了组内排序策略。版本号为Hadoop1.1.2