一：背景

Reduce端連線比Map端連線更為普遍，因為輸入的資料不需要特定的結構，但是效率比較低，因為所有資料都必須經過Shuffle過程。

二：技術實現

基本思路

(1)：Map端讀取所有的檔案，並在輸出的內容里加上標示，代表資料是從哪個檔案裡來的。

(2)：在reduce處理函式中，按照標識對資料進行處理。

(3)：然後根據Key去join來求出結果直接輸出。

資料準備

準備好下面兩張表：

(1)：tb_a(以下簡稱表A)

id	name
1	北京
2	天津
3	河北
4	山西
5	內蒙古
6	遼寧
7	吉林
8	黑龍江

(2)：tb_b(以下簡稱表B)

id	statyear	num
1	2010	1962
1	2011	2019
2	2010	1299
2	2011	1355
4	2011	3574
4	2011	3593
9	2010	2303
9	2011	2347
 

計算模型

整個計算過程是：

(1)：在Map階段，把所有資料標記成<key,value>的形式，其中key是id，value則根據來源不同取不同的形式：來源於A的記錄，value的值為"a#"+name；來源於B的記錄，value的值為"b#"+score。

(2)：在reduce階段，先把每個key下的value列表拆分為分別來自表A和表B的兩部分，分別放入兩個向量中。然後遍歷兩個向量做笛卡爾積，形成一條條最終的結果。

如下圖所示：

程式碼實現如下：

public class ReduceJoinTest {

		// 定義輸入路徑
		private static final String INPUT_PATH = "hdfs://liaozhongmin:9000/table_join/tb_*";
		// 定義輸出路徑
		private static final String OUT_PATH = "hdfs://liaozhongmin:9000/out";

		public static void main(String[] args) {

			try {
				// 建立配置資訊
				Configuration conf = new Configuration();
				

				// 建立檔案系統
				FileSystem fileSystem = FileSystem.get(new URI(OUT_PATH), conf);
				// 如果輸出目錄存在，我們就刪除
				if (fileSystem.exists(new Path(OUT_PATH))) {
					fileSystem.delete(new Path(OUT_PATH), true);
				}

				// 建立任務
				Job job = new Job(conf, ReduceJoinTest.class.getName());

				//1.1	設定輸入目錄和設定輸入資料格式化的類
				FileInputFormat.setInputPaths(job, INPUT_PATH);
				job.setInputFormatClass(TextInputFormat.class);

				//1.2	設定自定義Mapper類和設定map函式輸出資料的key和value的型別
				job.setMapperClass(ReduceJoinMapper.class);
				job.setMapOutputKeyClass(Text.class);
				job.setMapOutputValueClass(Text.class);

				//1.3	設定分割槽和reduce數量(reduce的數量，和分割槽的數量對應，因為分割槽為一個，所以reduce的數量也是一個)
				job.setPartitionerClass(HashPartitioner.class);
				job.setNumReduceTasks(1);

				//1.4	排序
				//1.5	歸約
				//2.1	Shuffle把資料從Map端拷貝到Reduce端。
				//2.2	指定Reducer類和輸出key和value的型別
				job.setReducerClass(ReduceJoinReducer.class);
				job.setOutputKeyClass(Text.class);
				job.setOutputValueClass(Text.class);

				//2.3	指定輸出的路徑和設定輸出的格式化類
				FileOutputFormat.setOutputPath(job, new Path(OUT_PATH));
				job.setOutputFormatClass(TextOutputFormat.class);


				// 提交作業 退出
				System.exit(job.waitForCompletion(true) ? 0 : 1);
			
			} catch (Exception e) {
				e.printStackTrace();
			}
		}
		
	public static  class ReduceJoinMapper extends Mapper<LongWritable, Text, Text, Text>{
		@Override
		protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context) throws IOException, InterruptedException {
			  //獲取輸入檔案的全路徑和名稱
			  FileSplit fileSplit = (FileSplit) context.getInputSplit();
			  String path = fileSplit.getPath().toString();
			  
			  //獲取輸入記錄的字串
			  String line = value.toString();
			  
			  //拋棄空記錄
			  if (line == null || line.equals("")){
				  return;
			  }
			  
			  //處理來自tb_a表的記錄
			  if (path.contains("tb_a")){
				  //按製表符切割
				  String[] values = line.split("\t");
				  //當陣列長度小於2時，視為無效記錄
				  if (values.length < 2){
					  return;
				  }
				  //獲取id和name
				  String id = values[0];
				  String name = values[1];
				  
				  //把結果寫出去
				  context.write(new Text(id), new Text("a#" + name));
			  } else if (path.contains("tb_b")){
				  //按製表符切割
				  String[] values = line.split("\t");
				  //當長度不為3時，視為無效記錄
				  if (values.length < 3){
					  return;
				  }
				  
				  //獲取屬性
				  String id = values[0];
				  String statyear = values[1];
				  String num = values[2];
				  
				  //寫出去
				  context.write(new Text(id), new Text("b#" + statyear + "	" + num));
			  }
			
		}
		
		public static class ReduceJoinReducer extends Reducer<Text, Text, Text, Text>{
			
			@Override
			protected void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, Text>.Context context) throws IOException, InterruptedException {
				
				//用來存放來自tb_a表的資料
				Vector<String> vectorA = new Vector<String>();
				//用來存放來自tb_b表的
				Vector<String> vectorB = new Vector<String>();
				
				//迭代集合資料
				for (Text val : values){
					//將集合中的資料對應新增到Vector中
					if (val.toString().startsWith("a#")){
						vectorA.add(val.toString().substring(2));
					} else if (val.toString().startsWith("b#")){
						vectorB.add(val.toString().substring(2));
					}
				}
				
				//獲取兩個Vector集合的長度
				int sizeA = vectorA.size();
				int sizeB = vectorB.size();
				
				//遍歷兩個向量將結果寫出去
				for (int i=0; i<sizeA; i++){
					for (int j=0; j<sizeB; j++){
						context.write(key, new Text("	" + vectorA.get(i) + "	" + vectorB.get(j)));
					}
				}
				
				
			}
		}
	}
}
 

細節：

(1)：當map讀取原始檔時，如何區分出是file1還是file2？

FileSplit fileSplit = (FileSplit)context.getInputSplit();
        String path =  fileSplit.getPath().toString();