hadoop 多表关联

package test;

import java.util.ArrayList;
import java.util.List;

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

public class MTjoin {

	public static class Map extends Mapper<LongWritable, Text, Text, Text>{
		private static Text k = new Text();
		private static Text v = new Text();
		
		protected void map(LongWritable key, Text value, Context context) 
				throws java.io.IOException ,InterruptedException {
			String[] splits = value.toString().split("\t");
			if(splits.length != 2){
				return ;
			}
			
			//取得文件名 a.txt(工厂名字，序号) b.txt(序号，地址)
			String fileName = ((FileSplit)context.getInputSplit()).getPath().getName();
			if("a.txt".equals(fileName)){
				k.set(splits[1]);
				v.set("1"+splits[0]);
			}else if("b.txt".equals(fileName)){
				k.set(splits[0]);
				v.set("2"+splits[1]);
			}else{
				return ;
			}
			context.write(k, v);
		};
	}
	public static class Reduce extends Reducer<Text, Text, Text, Text>{
		private static List<String> names = new ArrayList<String>();
		private static List<String> addrs = new ArrayList<String>();
		private static Text name = new Text();
		private static Text addr = new Text();
		
		protected void reduce(Text key, Iterable<Text> values, Context context) 
				throws java.io.IOException ,InterruptedException {
			for (Text value : values) {
				String temp = value.toString();
				if(temp.startsWith("1")){
					names.add(temp.substring(1));
				}else{
					addrs.add(temp.substring(1));
				}
			}
			for (String n : names) {
				for (String a : addrs) {
					name.set(n);
					addr.set(a);
					context.write(name, addr);
				}
			}
			names.clear();
			addrs.clear();
		};
	}
	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		String[] otherArgs = new GenericOptionsParser(conf,args).getRemainingArgs();
		if(otherArgs.length != 2){
			System.err.println("Usage:MTjoin");
			System.exit(2);
		}
		Job job = new Job(conf, "MTjoin");
		job.setJarByClass(MTjoin.class);
		
		job.setMapperClass(Map.class);
		job.setReducerClass(Reduce.class);
		
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Text.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		
		FileInputFormat.addInputPath(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));
		
		System.exit(job.waitForCompletion(true) ? 0 : 1);
		
	}
}