我们都知道在数据库里，多个表之间是可以根据某个链接键进行join的，这也是数据库的范式规范，通过主外键的关联，由此来减少数据冗余，提高性能。当然近几年，随着NOSQL的兴起，出现了基于列的的列式数据库，典型的有Hbase，MongonDB，Cassdran，等等，NOSQL数据库弱化了关联，直接将一整条数据，存入一列，以及去掉了数据库的部分事务特性，从而在海量数据面前显得游刃有余，当然，大部分的NOSQL不支持join操作，也没有绝对的必要支持，因为现在，我们完全是把一整条数据存在了一起，虽然多了许多冗余，但也换来了比较高检索性能，扩展性能，可靠性能。但某些业务场景下，我们仍然需要Join操作，这时候怎么办？

a文件的数据1,三劫散仙,135754682482,凤舞九天,189652358743,忙忙碌碌,159868547894,少林寺方丈,15698745862

b文件的数据3,A,99,2013-03-051,B,89,2013-02-052,C,69,2013-03-093,D,56,2013-06-07

package com.qin.reducejoin;import java.io.DataInput;  import java.io.DataOutput;  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.FileSystem;import org.apache.hadoop.fs.Path;  import org.apache.hadoop.io.Text;  import org.apache.hadoop.io.Writable;  import org.apache.hadoop.mapred.FileInputFormat;  import org.apache.hadoop.mapred.FileOutputFormat;  import org.apache.hadoop.mapred.JobClient;  import org.apache.hadoop.mapred.JobConf;  import org.apache.hadoop.mapred.KeyValueTextInputFormat;  import org.apache.hadoop.mapred.MapReduceBase;  import org.apache.hadoop.mapred.Mapper;  import org.apache.hadoop.mapred.OutputCollector;  import org.apache.hadoop.mapred.Reducer;  import org.apache.hadoop.mapred.Reporter;  import org.apache.hadoop.mapred.TextInputFormat;  import org.apache.hadoop.mapred.TextOutputFormat;  import org.apache.hadoop.util.ReflectionUtils;import org.apache.hadoop.util.Tool;  import org.apache.hadoop.util.ToolRunner;    import org.apache.hadoop.contrib.utils.join.DataJoinMapperBase;  import org.apache.hadoop.contrib.utils.join.DataJoinReducerBase;  import org.apache.hadoop.contrib.utils.join.TaggedMapOutput;  import com.qin.joinreduceerror.JoinReduce;  /*** *  * Hadoop1.2的版本，旧版本实现的Reduce侧连接 *  * @author qindongliang *  *    大数据交流群：376932160 *  搜索技术交流群：324714439 *  *  */public class DataJoin extends Configured implements Tool {        		/**	 * 	 * Map实现	 * 	 * */    public static class MapClass extends DataJoinMapperBase {                	/**    	 * 读取输入的文件路径    	 *     	 * **/        protected Text generateInputTag(String inputFile) {                      	//返回文件路径，做标记            return new Text(inputFile);          }                            /***         * 分组的Key         *          * **/                protected Text generateGroupKey(TaggedMapOutput aRecord) {              String line = ((Text) aRecord.getData()).toString();              String[] tokens = line.split(",");              String groupKey = tokens[0];              return new Text(groupKey);          }                                            protected TaggedMapOutput generateTaggedMapOutput(Object value) {              TaggedWritable retv = new TaggedWritable((Text) value);              retv.setTag(this.inputTag);              return retv;          }      }            /**     *      * Reduce进行笛卡尔积     *      * **/    public static class Reduce extends DataJoinReducerBase {                	    	/***    	 * 笛卡尔积    	 *     	 * */        protected TaggedMapOutput combine(Object[] tags, Object[] values) {              if (tags.length < 2) return null;                String joinedStr = "";               for (int i=0; i
    
      0) {joinedStr += ",";}                  TaggedWritable tw = (TaggedWritable) values[i];                  String line = ((Text) tw.getData()).toString();                  String[] tokens = line.split(",", 2);                  joinedStr += tokens[1];              }              TaggedWritable retv = new TaggedWritable(new Text(joinedStr));              retv.setTag((Text) tags[0]);               return retv;          }      }            /**     *      * 自定义的输出类型     *      * ***/    public static class TaggedWritable extends TaggedMapOutput {                private Writable data;                    /**         * 注意加上构造方法         *          * */        public TaggedWritable() {			// TODO Auto-generated constructor stub		}                public TaggedWritable(Writable data) {              this.tag = new Text("");              this.data = data;          }                    public Writable getData() {              return data;          }                    public void write(DataOutput out) throws IOException {              this.tag.write(out);              //此行代码很重要            out.writeUTF(this.data.getClass().getName());                       this.data.write(out);                     }                    public void readFields(DataInput in) throws IOException {              this.tag.readFields(in);                //加入此部分代码，否则，可能报空指针异常        	String temp=in.readUTF();			if (this.data == null|| !this.data.getClass().getName().equals(temp)) {					try {					this.data = (Writable) ReflectionUtils.newInstance(					Class.forName(temp), null);					} catch (ClassNotFoundException e) {					e.printStackTrace();					}					}            this.data.readFields(in);          }      }            public int run(String[] args) throws Exception {          Configuration conf = getConf();                    JobConf job = new JobConf(conf, DataJoin.class);                  job.set("mapred.job.tracker","192.168.75.130:9001");      		读取person中的数据字段      		 job.setJar("tt.jar");      	job.setJarByClass(DataJoin.class);    		System.out.println("模式：  "+job.get("mapred.job.tracker"));;      		       		       		       		         String path="hdfs://192.168.75.130:9000/root/outputjoindb";		FileSystem fs=FileSystem.get(conf);		Path p=new Path(path);		if(fs.exists(p)){			fs.delete(p, true);			System.out.println("输出路径存在，已删除！");		}	                         Path in = new Path("hdfs://192.168.75.130:9000/root/inputjoindb");        //  Path out = new Path(args[1]);          FileInputFormat.setInputPaths(job, in);          FileOutputFormat.setOutputPath(job, p);                    job.setJobName("cee");          job.setMapperClass(MapClass.class);          job.setReducerClass(Reduce.class);                    job.setInputFormat(TextInputFormat.class);          job.setOutputFormat(TextOutputFormat.class);          job.setOutputKeyClass(Text.class);          job.setOutputValueClass(TaggedWritable.class);          job.set("mapred.textoutputformat.separator", ",");                    JobClient.runJob(job);           return 0;      }            public static void main(String[] args) throws Exception {       	    	    	        int res = ToolRunner.run(new Configuration(),                                   new DataJoin(),                                   args);                    System.exit(res);      }  }
    

模式：  192.168.75.130:9001输出路径存在，已删除！INFO - NativeCodeLoader.
    
     (43) | Loaded the native-hadoop libraryWARN - LoadSnappy.
     
      (46) | Snappy native library not loadedINFO - FileInputFormat.listStatus(199) | Total input paths to process : 2INFO - JobClient.monitorAndPrintJob(1380) | Running job: job_201404222310_0025INFO - JobClient.monitorAndPrintJob(1393) |  map 0% reduce 0%INFO - JobClient.monitorAndPrintJob(1393) |  map 33% reduce 0%INFO - JobClient.monitorAndPrintJob(1393) |  map 100% reduce 0%INFO - JobClient.monitorAndPrintJob(1393) |  map 100% reduce 33%INFO - JobClient.monitorAndPrintJob(1393) |  map 100% reduce 100%INFO - JobClient.monitorAndPrintJob(1448) | Job complete: job_201404222310_0025INFO - Counters.log(585) | Counters: 30INFO - Counters.log(587) |   Job Counters INFO - Counters.log(589) |     Launched reduce tasks=1INFO - Counters.log(589) |     SLOTS_MILLIS_MAPS=14335INFO - Counters.log(589) |     Total time spent by all reduces waiting after reserving slots (ms)=0INFO - Counters.log(589) |     Total time spent by all maps waiting after reserving slots (ms)=0INFO - Counters.log(589) |     Launched map tasks=3INFO - Counters.log(589) |     Data-local map tasks=3INFO - Counters.log(589) |     SLOTS_MILLIS_REDUCES=9868INFO - Counters.log(587) |   File Input Format Counters INFO - Counters.log(589) |     Bytes Read=207INFO - Counters.log(587) |   File Output Format Counters INFO - Counters.log(589) |     Bytes Written=172INFO - Counters.log(587) |   FileSystemCountersINFO - Counters.log(589) |     FILE_BYTES_READ=837INFO - Counters.log(589) |     HDFS_BYTES_READ=513INFO - Counters.log(589) |     FILE_BYTES_WRITTEN=221032INFO - Counters.log(589) |     HDFS_BYTES_WRITTEN=172INFO - Counters.log(587) |   Map-Reduce FrameworkINFO - Counters.log(589) |     Map output materialized bytes=849INFO - Counters.log(589) |     Map input records=8INFO - Counters.log(589) |     Reduce shuffle bytes=849INFO - Counters.log(589) |     Spilled Records=16INFO - Counters.log(589) |     Map output bytes=815INFO - Counters.log(589) |     Total committed heap usage (bytes)=496644096INFO - Counters.log(589) |     CPU time spent (ms)=2080INFO - Counters.log(589) |     Map input bytes=187INFO - Counters.log(589) |     SPLIT_RAW_BYTES=306INFO - Counters.log(589) |     Combine input records=0INFO - Counters.log(589) |     Reduce input records=8INFO - Counters.log(589) |     Reduce input groups=4INFO - Counters.log(589) |     Combine output records=0INFO - Counters.log(589) |     Physical memory (bytes) snapshot=623570944INFO - Counters.log(589) |     Reduce output records=4INFO - Counters.log(589) |     Virtual memory (bytes) snapshot=2908262400INFO - Counters.log(589) |     Map output records=8