輸入檔案1：

factoryname 　　addressed
Beijing Red Star 1
Shenzhen Thunder 3
Guangzhou Honda 2
Beijing Rising 1
Guangzhou Development Bank 2
Tencent 3
Back of Beijing 1

輸入檔案2：

addressID    addressname
1 Beijing
2 Guangzhou
3 Shenzhen
4 Xian

要想得到的輸出檔案：

factoryname	addressname
Beijing Red Star                	Beijing 
Beijing Rising                   　　　 	Beijing 
Back of Beijing                　　　　 	Beijing 
Guangzhou Honda            　　　       	Guangzhou 
Guangzhou Development Bank             	Guangzhou 
Shenzhen Thunder            　　　      	Shenzhen 
Tencent                　　　　　　　  	Shenzhen 
 

自己編寫的執行正確的map/reduce原始碼：

package bin;

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.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.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;

public class MTJoin {
	
//	對於兩個不同的輸入檔案的處理，前面的檔案都是相同格式的，這裡的兩個檔案格式不同，實現的是兩個檔案的自然連線
	public static class MTJoinMap extends Mapper<Object, Text, Text, Text>{
		//map做的事是將不同檔案中的資料進行處理分發；繼承了mapper之後就要實現其中的map方法
		public void map(Object key,Text value,Context context) {
			//每次執行map函式，就會讀取檔案中的一行資料給value，要對value進行處理；
			String line=new String(value.toString());
			String MapKey=new String();
			String MapValue=new String();
			
			if (line.contains("factoryname")||line.contains("addressed")||line.contains("addressID")||line.contains("addressname")) {
				return;
			}
			int len=line.length();
			String flag=new String();//記錄屬於左表還是屬於右表；
			//每個map任務按行讀取檔案，將讀取的一行資料交給map函式執行處理，即，每一行資料呼叫一次map函式，
			//所以這裡的一個tokenizer的容器，包含的資料，在這裡只有兩個，當讀取下一行資料時，就會再次呼叫一次MTJoinMap的map函式
			//------------------------------由此，這裡的i的意義就很明顯了。
			//當每一行資料讀進來初始時，i的值均為0,所以當檔案為第一個檔案，即第二位為ID時，flag="1"，當檔案為第一個檔案時，即第一位為ID,則，i=0,--> flag="2"
			String[] record=line.split(" ");
			if (len==0) {
				return;
			}else {
				if (line.charAt(0) > '0' && line.charAt(0) < '9') {
					flag="2";
					MapKey=String.valueOf(line.charAt(0));
					for (int i = 1; i < record.length; i++) {
						MapValue =MapValue.concat(record[i]).concat(" ");
					}
					
				}else if (line.charAt(len-1) > '0' && line.charAt(len-1) < '9') {
					flag="1";
					MapKey=String.valueOf(line.charAt(len-1));
					for (int i = 0; i < record.length-1; i++) {
						if (record[i].equals(" ")) {
							continue;
						}
						MapValue =MapValue.concat(record[i]).concat(" ");
					}
				}
				try {
					context.write(new Text(MapKey), new Text(flag+"+"+MapValue));
				} catch (IOException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}
			
		}
		
	}
	public static int time=0;
	
	public static class MTJoinReduce extends Reducer<Text, Text, Text, Text>{
		public void reduce(Text key,Iterable<Text> values,Context context) {
			
			if (time==0) {
				try {
					context.write(new Text("factoryname"), new Text("addressname"));
				} catch (IOException | InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
                time++; 
			}//輸出表頭資訊factory name address name
			
			String[]  factoryname=new String[10];
			int factoryNum=0;
			String addressname=new String();
			
			Iterator<Text> iterator=values.iterator();
			while (iterator.hasNext()) {
				String record= iterator.next().toString();
				
				int len=record.length();//以下這三句不加會報錯： Exception closing file /user/xinxin/output/MutiTableJoin_out
				if (len==0) {//一定要判定迭代器裡面的字串為空的情況
					continue;
				}
				char flag=record.charAt(0);
				
				if (flag=='1' && record.substring(2).length()!=0) {
					
					factoryname[factoryNum]=record.substring(2);
					factoryNum++;
				}else if (flag=='2' && record.substring(2).length()!=0) 
					addressname = record.substring(2);
			}
			if (factoryNum!=0 && addressname.length()!=0) {
				for (int i = 0; i < factoryNum; i++) {
					try {
						context.write(new Text(factoryname[i]), new Text(addressname));
					} catch (IOException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					} catch (InterruptedException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
				}
			}
		}
	}
	
	
	public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
		Configuration configuration=new Configuration();
		String[] otherArgs = new GenericOptionsParser(configuration,args).getRemainingArgs();
		 if (otherArgs.length !=2) {
			System.err.println("Usage: MTJoin <in> <out>");
			System.exit(2);
		}
		 Job job =new Job(configuration, "tracert MTJoin");
		 job.setJarByClass(MTJoin.class);
		 
		 job.setMapperClass(MTJoinMap.class);
		 job.setReducerClass(MTJoinReduce.class);
		 job.setOutputKeyClass(Text.class);
		 job.setOutputValueClass(Text.class);
		 
		 FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
		 FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
		 
		 System.exit(job.waitForCompletion(true)? 0 : 1);
	}
}
 

作者原創，如果轉載請註明出處，謝謝

---------------------------------------------------------欣