MapReduce 中的两表 join 实例(二)
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 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 package com.baidu.uilt; import java.io.*; import org.apache.hadoop.io.*; public class TextPair implements WritableComparable<TextPair>{ private Textfirst; private Textsecond; public TextPair(){ set( new Text(), new Text()); } public TextPair(Stringfirst,Stringsecond){ set( new Text(first), new Text(second)); } public TextPair(Textfirst,Textsecond){ set(first,second); } public void set(Textfirst,Textsecond){ this .first=first; this .second=second; } public TextgetFirst(){ return first; } public TextgetSecond(){ return second; } @Override public void write(DataOutputout) throws IOException{ first.write(out); second.write(out); } @Override public void readFields(DataInputin) throws IOException{ first.readFields(in); second.readFields(in); } @Override public int hashCode(){ return first.hashCode()* 163 +second.hashCode(); } @Override public boolean equals(Objecto){ if (o instanceof TextPair){ TextPairtp=(TextPair)o; return first.equals(tp.first)&&second.equals(tp.second); } return false ; } @Override public StringtoString(){ return first+ "\t" +second; } @Override public int compareTo(TextPairtp){ int cmp=first.compareTo(tp.first); if (cmp!= 0 ){ return cmp; } return second.compareTo(tp.second); } public static class Comparator extends WritableComparator{ private static final Text.ComparatorTEXT_COMPARATOR= new Text.Comparator(); public Comparator(){ super (TextPair. class ); } @Override public int compare( byte []b1, int s1, int l1, byte []b2, int s2, int l2){ try { int firstL1=WritableUtils.decodeVIntSize(b1[s1])+readVInt(b1,s1); int firstL2=WritableUtils.decodeVIntSize(b2[s2])+readVInt(b2,s2); int cmp=TEXT_COMPARATOR.compare(b1,s1,firstL1,b2,s2,firstL2); if (cmp!= 0 ){ return cmp; } return TEXT_COMPARATOR.compare(b1,s1+firstL1,l1-firstL1, b2,s2+firstL2,l2-firstL2); } catch (IOExceptione){ throw new IllegalArgumentException(e); } } } static { WritableComparator.define(TextPair. class , new Comparator()); } public static class FirstComparator extends WritableComparator{ private static final Text.ComparatorTEXT_COMPARATOR= new Text.Comparator(); public FirstComparator(){ super (TextPair. class ); } @Override public int compare( byte []b1, int s1, int l1, byte []b2, int s2, int l2){ try { int firstL1=WritableUtils.decodeVIntSize(b1[s1])+readVInt(b1,s1); int firstL2=WritableUtils.decodeVIntSize(b2[s2])+readVInt(b2,s2); return TEXT_COMPARATOR.compare(b1,s1,firstL1,b2,s2,firstL2); } catch (IOExceptione){ throw new IllegalArgumentException(e); } } @Override public int compare(WritableComparablea,WritableComparableb){ if (a instanceof TextPair&&b instanceof TextPair){ return ((TextPair)a).first.compareTo(((TextPair)b).first); } return super .compare(a,b); } } } 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 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 package com.baidu.loan; /*** * */home/users/ouerqiang/hadoop/hadoop-client-palo/hadoop/bin/hadoopjarLoanIdeaInfoText.jarcom.baidu.loan.LoanIdeainfoJoinIterialByDAILI6/test/fbiz/loan/ideainfo/LoanIdeainfoByDAILIUnitID_0928/test/fbiz/loan/ideainfo/LoanIterialByDAI_0928/test/fbiz/loan/ideainfo/LoanIdeainfoJoinIterialByDAILI6_1_0928 * ***/ import java.io.IOException; import java.util.Iterator; import org.apache.hadoop.mapred.FileOutputFormat; import org.apache.hadoop.mapred.JobClient; import org.apache.hadoop.mapred.JobConf; import org.apache.hadoop.mapred.MapReduceBase; import org.apache.hadoop.mapred.Mapper; import org.apache.hadoop.mapred.OutputCollector; import org.apache.hadoop.mapred.Partitioner; import org.apache.hadoop.mapred.Reducer; import org.apache.hadoop.mapred.Reporter; import org.apache.hadoop.mapred.TextInputFormat; import org.apache.hadoop.mapred.lib.MultipleInputs; import org.apache.hadoop.util.Tool; import org.apache.hadoop.util.ToolRunner; 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 com.baidu.uilt.TextPair; public class LoanIdeainfoJoinIterialByDAILI6 extends Configured implements Tool{ public static class JoinUnitMapper extends MapReduceBase implements Mapper<LongWritable,Text,TextPair,Text>{ public void map(LongWritablekey,Textvalue, OutputCollector<TextPair,Text>output,Reporterreporter) throws IOException{ StringgbkStr=value.toString(); if (gbkStr.split( "\t" ).length< 2 &&gbkStr.split( "," ).length== 4 ){ String[]strs=gbkStr.split( "," ); output.collect( new TextPair(strs[ 0 ], "0" ),value); } } } public static class JoinIterialMapper extends MapReduceBase implements Mapper<LongWritable,Text,TextPair,Text>{ public void map(LongWritablekey,Textvalue, OutputCollector<TextPair,Text>output,Reporterreporter) throws IOException{ StringgbkStr=value.toString(); if (gbkStr.split( "\t" ).length> 4 ){ //LoanIterial String[]strs=gbkStr.split( "\t" ); output.collect( new TextPair(strs[ 0 ], "1" ),value); } } } public static class JoinReducer extends MapReduceBase implements Reducer<TextPair,Text,Text,Text>{ public void reduce(TextPairkey,Iterator<Text>values, OutputCollector<Text,Text>output,Reporterreporter) throws IOException{ TextstationName= new Text(values.next()); while (values.hasNext()){ Textrecord=values.next(); TextoutValue= new Text(stationName.toString()+ "\t" +record.toString()); output.collect(stationName,record); //output.collect(key.getFirst(),outValue); } } } public static class KeyPartitioner implements Partitioner<TextPair,Text>{ @Override public void configure(JobConfjob){} @Override public int getPartition(TextPairkey,Textvalue, int numPartitions){ return (key.getFirst().hashCode()&Integer.MAX_VALUE)%numPartitions; } } @Override public int run(String[]args) throws Exception{ if (args.length!= 3 ){ return - 1 ; } JobConfconf= new JobConf(getConf(),getClass()); conf.setJobName( "Joinrecordwithstationname" ); StringstrPathUnit=args[ 0 ]; StringstrPathIterial=args[ 1 ]; PathoutputPath= new Path(args[ 2 ]); MultipleInputs.addInputPath(conf, new Path(strPathUnit), TextInputFormat. class ,JoinUnitMapper. class ); MultipleInputs.addInputPath(conf, new Path(strPathIterial), TextInputFormat. class ,JoinIterialMapper. class ); FileOutputFormat.setOutputPath(conf,outputPath); conf.setPartitionerClass(KeyPartitioner. class ); conf.setOutputValueGroupingComparator(TextPair.FirstComparator. class ); conf.setMapOutputKeyClass(TextPair. class ); conf.setReducerClass(JoinReducer. class ); conf.setOutputKeyClass(Text. class ); JobClient.runJob(conf); return 0 ; } public static void main(String[]args) throws Exception{ int exitCode=ToolRunner.run( new LoanIdeainfoJoinIterialByDAILI6(),args); System.exit(exitCode); } } 需要注意的是上面的代码只是针对两表的一对一（多）关系，不满足多对多关系。如果需要满足多对多关系则需要加上一下判断即可。 本文转自 梦朝思夕 51CTO博客，原文链接:http://blog.51cto.com/qiangmzsx/1560553
