import java.io.DataOutput;import java.io.IOException;import java.util.HashMap;import java.util.Map;import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.conf.Configured;import org.apache.hadoop.fs.FSDataOutputStream;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.mapreduce.InputSplit;import org.apache.hadoop.mapreduce.Job;import org.apache.hadoop.mapreduce.JobContext;import org.apache.hadoop.mapreduce.Mapper;import org.apache.hadoop.mapreduce.RecordReader;import org.apache.hadoop.mapreduce.RecordWriter;import org.apache.hadoop.mapreduce.Reducer;import org.apache.hadoop.mapreduce.Reducer.Context;import org.apache.hadoop.mapreduce.TaskAttemptContext;import org.apache.hadoop.mapreduce.lib.input.CombineFileInputFormat;import org.apache.hadoop.mapreduce.lib.input.CombineFileRecordReader;import org.apache.hadoop.mapreduce.lib.input.CombineFileSplit;import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;import org.apache.hadoop.mapreduce.lib.input.FileSplit;import org.apache.hadoop.mapreduce.lib.input.LineRecordReader;import org.apache.hadoop.mapreduce.lib.input.SequenceFileRecordReader;import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;import org.apache.hadoop.util.ReflectionUtils;import org.apache.hadoop.util.Tool;import org.apache.hadoop.util.ToolRunner;public class TestCombine extends Configured implements Tool {	private static class ProvinceMapper extends			Mapper<Object, Text, Text, Text> {		@Override		protected void map(Object key, Text value, Context context)				throws IOException, InterruptedException {			System.out.println("value : " + value + " Context " + context);			context.write(value, value);		}	}	private static class ProvinceReducer extends			Reducer<Text, Text, Text, Text> {		@Override		protected void reduce(Text key, Iterable<Text> values, Context context)				throws IOException, InterruptedException {			for (Text va : values) {			    System.out.println("reduce " + key);				context.write(key, key);			}		}	}		 // 输入格式     static class CombineSequenceFileInputFormat<K, V> extends CombineFileInputFormat<K, V> {  	    @SuppressWarnings({ "unchecked", "rawtypes" })  	    @Override  	    public RecordReader<K, V> createRecordReader(InputSplit split, TaskAttemptContext context) throws IOException {  	        return new CombineFileRecordReader((CombineFileSplit)split, context, CombineLineRecordReader.class);  	    }  	}  		 static class CombineLineRecordReader<K, V> extends RecordReader<K, V> {  	    private CombineFileSplit split;  	    private TaskAttemptContext context;  	    private int index;  	    private RecordReader<K, V> rr;  	  	    @SuppressWarnings("unchecked")  	    public CombineLineRecordReader(CombineFileSplit split, TaskAttemptContext context, Integer index) throws IOException, InterruptedException {  	        this.index = index;	        this.split = (CombineFileSplit) split;  	        this.context = context;  	  	        this.rr = (RecordReader<K, V>) ReflectionUtils.newInstance(LineRecordReader.class, context.getConfiguration());  	    }  	  	    @SuppressWarnings("unchecked")  	    @Override  	    public void initialize(InputSplit curSplit, TaskAttemptContext curContext) throws IOException, InterruptedException {  	        this.split = (CombineFileSplit) curSplit;  	        this.context = curContext;  	  	        if (null == rr) {  	            rr = ReflectionUtils.newInstance(SequenceFileRecordReader.class, context.getConfiguration());  	        }  	  	        FileSplit fileSplit = new FileSplit(this.split.getPath(index),  	                this.split.getOffset(index), this.split.getLength(index),  	                this.split.getLocations());  	          	        this.rr.initialize(fileSplit, this.context);  	    }  	  	    @Override  	    public float getProgress() throws IOException, InterruptedException {  	        return rr.getProgress();  	    }  	  	    @Override  	    public void close() throws IOException {  	        if (null != rr) {  	            rr.close();  	            rr = null;  	        }  	    }  	  	    @Override  	    public K getCurrentKey()  	    throws IOException, InterruptedException {  	        return rr.getCurrentKey();  	    }  	  	    @Override  	    public V getCurrentValue()  	    throws IOException, InterruptedException {  	        return rr.getCurrentValue();  	    }  	  	    @Override  	    public boolean nextKeyValue() throws IOException, InterruptedException {  	        return rr.nextKeyValue();  	    }  	}  		// 输出格式	 static class MyOutputFormat extends FileOutputFormat<Text, Text>{		@Override		public RecordWriter<Text, Text> getRecordWriter(				TaskAttemptContext job) throws IOException, InterruptedException {			return new MyRecordWriter(job);		}	}	  public static class  MyRecordWriter extends RecordWriter<Text, Text> {				private Map<String, FSDataOutputStream> outputMap = null;		private static final String LINESEPARATOR = "\n";		private FileSystem fs;		private JobContext job;				public MyRecordWriter(JobContext job) throws IOException {			this.outputMap = new HashMap<String, FSDataOutputStream>();			this.job = job;			this.fs = FileSystem.get(job.getConfiguration());		}				// 参考 MultipleOutputs		public void write(Text key, Text value) throws IOException {			String k = key.toString();			if(k.isEmpty())				return;			FSDataOutputStream out = outputMap.get(k);			if(out==null) {				if(k.isEmpty())					System.out.println(value.toString());				Path outputPath = new Path(FileOutputFormat.getOutputPath(job), k);				if(!fs.exists(outputPath))					out = fs.create(outputPath);				else					return;				outputMap.put(k, out);			}			out.write(value.getBytes());			out.write(LINESEPARATOR.getBytes());		}		@Override		public void close(TaskAttemptContext context) throws IOException,				InterruptedException {			for(FSDataOutputStream out : outputMap.values()) {				out.close();			}		}	}		public int run(String[] args) throws Exception {		Configuration conf = new Configuration();				Job job = new Job(conf);		job.setJobName("TestCombine");		job.setJarByClass(TestCombine.class);		job.setMapperClass(ProvinceMapper.class);		job.setReducerClass(ProvinceReducer.class);				//job.setInputFormatClass(CombineSequenceFileInputFormat.class);		job.setOutputFormatClass(MyOutputFormat.class);				job.setOutputKeyClass(Text.class);		job.setOutputValueClass(Text.class);				String inpath = "/home/hadoop/tmp/combine";		String outpath = "/home/hadoop/tmp/combineout";		Path p = new Path(outpath);				FileSystem fs = FileSystem.get(conf);		if (fs.exists(p)){			fs.delete(p);		}		FileInputFormat.addInputPaths(job, inpath);		FileOutputFormat.setOutputPath(job, p);		return job.waitForCompletion(true) ? 0 : 1;	} 	public static void main(String[] args) throws Exception {		int ret = ToolRunner.run(new TestCombine(), args);		System.exit(ret);	} }