Hadoop的MapReduce中多文件输出

这两天在网上看了个MapReduce的多文件输出的帖子： http://blog.csdn.net/inkfish。写的不错。

我试着完成了一下。也是分为三个文件：我这三个文件，跟原作者的稍有不同。其中有些类是我原来写的，我直接拷贝过来的，所以有点不同。

My_LineRead.java

public class My_LineRead<K, V> extends RecordWriter<K, V>{
		private static final String utf8 = "UTF-8";
		private static final  String colon = "----";  //划分符号
	    private static final byte[] newline;
	    static {
	      try {
	        newline = "/n".getBytes(utf8);
	      } catch (UnsupportedEncodingException uee) {
	        throw new IllegalArgumentException("can't find " + utf8 + " encoding");
	      }
	    }
	    protected DataOutputStream out;
	    private final byte[] keyValueSeparator;
	    
	    public My_LineRead(DataOutputStream out) {
	        this(out, colon); //调用下面的构造函数
	    }
		public My_LineRead(DataOutputStream out, String keyValueSeparator) {
			// TODO Auto-generated constructor stub
			this.out = out;
			try {
				this.keyValueSeparator = keyValueSeparator.getBytes(utf8);
			} catch (UnsupportedEncodingException e) {
				// TODO Auto-generated catch block
				throw new IllegalArgumentException("can't find " + utf8 + " encoding");
			}
		}
		@Override
		public void close(TaskAttemptContext arg0) throws IOException,
				InterruptedException {
			// TODO Auto-generated method stub
			out.close();
		}

		@Override
		public void write(K key, V value) throws IOException,
				InterruptedException {
			if (!(key == null && key instanceof NullWritable)){
				//如果key不为空者输出key
				if ((Object)key instanceof Text){
					Text to = (Text) key;
					out.write(to.getBytes(), 0, to.getLength());
				}
				else
				{
					out.write(key.toString().getBytes(utf8));
				}
				out.write(keyValueSeparator);
			}
			if (!(value == null && value instanceof NullWritable)){
				//如果value不为空则输出value
				if ((Object)value instanceof Text){
					Text to = (Text) value;
					out.write(to.getBytes(), 0, to.getLength());
				}
				else
				{
					out.write(value.toString().getBytes(utf8));
				}
				out.write(newline);
			}
			
		}
	}

MyMultipleOutputFormat.java //这个类，我添加了些注释便于理解

public abstract class MyMultipleOutputFormat  <K extends WritableComparable<?>, V extends Writable>  
        extends FileOutputFormat<K, V> {
	//接口类，需要在主程序中实现generateFileNameForKeyValue来获取文件名
	private MultiRecordWriter writer = null;  
	@Override
	public RecordWriter<K, V> getRecordWriter(TaskAttemptContext job)
			throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		//如果第一次调用那么writer=null
		if (writer == null) {  
			//getTaskOutputPath获取output路径
			writer = new MultiRecordWriter(job, getTaskOutputPath(job));  
		}  
		return writer;
	}
	private Path getTaskOutputPath(TaskAttemptContext conf) throws IOException {
		Path workPath = null;
		OutputCommitter committer = super.getOutputCommitter(conf);
		if (committer instanceof FileOutputCommitter) {
			workPath = ((FileOutputCommitter) committer).getWorkPath();
		} else {
			Path outputPath = super.getOutputPath(conf);
			if (outputPath == null) {
				throw new IOException("Undefined job output-path");
			}
			workPath = outputPath;
		}
		return workPath;
	}
	/**通过key, value, conf来确定输出文件名（含扩展名）*/
	//返回值就是文件名。可以根据key,value来判断
	protected abstract String generateFileNameForKeyValue(K key, V value, Configuration conf);
	
	//MultiRecordWriter类
	public class MultiRecordWriter extends RecordWriter<K, V> {
		/**RecordWriter的缓存*/
		private HashMap<String, RecordWriter<K, V>> recordWriters = null;
		private TaskAttemptContext job = null;
		/**输出目录*/
		private Path workPath = null;
		//构造函数
		public MultiRecordWriter(TaskAttemptContext job, Path workPath) {
			super();
			this.job = job;
			this.workPath = workPath;
			recordWriters = new HashMap<String, RecordWriter<K, V>>();
		}
		//关闭，应该可能是多个文件进行关闭，所有采用循环
		//recordWriters.values() 就是指的getBaseRecordWriter返回的值。
		@Override
		public void close(TaskAttemptContext context) throws IOException, InterruptedException {
			Iterator<RecordWriter<K, V>> values = this.recordWriters.values().iterator();
			while (values.hasNext()) {
				values.next().close(context);
			}
			this.recordWriters.clear();
		}
		@Override
		public void write(K key, V value) throws IOException, InterruptedException {
			//得到输出文件名
			String baseName = generateFileNameForKeyValue(key, value, job.getConfiguration());
			//如果recordWriters里没有文件名，那么就建立。否则就直接写值。
			RecordWriter<K, V> rw = this.recordWriters.get(baseName);
			if (rw == null) {
				rw = getBaseRecordWriter(job, baseName);
				//放入HashMap
				this.recordWriters.put(baseName, rw);
			}
			rw.write(key, value);
		}
		// ${mapred.out.dir}/_temporary/_${taskid}/${nameWithExtension}
		private RecordWriter<K, V> getBaseRecordWriter(TaskAttemptContext job, String baseName)
				throws IOException, InterruptedException {
			//获取配置文件
			Configuration conf = job.getConfiguration();
			//查看是否使用解码器
			boolean isCompressed = getCompressOutput(job);
			String keyValueSeparator = ",";
			RecordWriter<K, V> recordWriter = null;
			if (isCompressed) {
				Class<? extends CompressionCodec> codecClass = getOutputCompressorClass(job,
						GzipCodec.class);
				CompressionCodec codec = ReflectionUtils.newInstance(codecClass, conf);
				Path file = new Path(workPath, baseName + codec.getDefaultExtension());
				FSDataOutputStream fileOut = file.getFileSystem(conf).create(file, false);
				recordWriter = new My_LineRead<K, V>(new DataOutputStream(codec
						.createOutputStream(fileOut)), keyValueSeparator);
			}
			//如果不使用解码器
			else {
				Path file = new Path(workPath, baseName);
				FSDataOutputStream fileOut = file.getFileSystem(conf).create(file, false);
				//recordWriter = new My_LineRead<K, V>(fileOut, keyValueSeparator);
				//这里我使用的我自己的OutputFormat
				recordWriter = new My_LineRead<K, V>(fileOut);
			}
			return recordWriter;
		}
	}
}

最后就是测试类，WordCount_MulFileOut.java

public class WordCount_MulFileOut {
	public static  class wordcountMapper extends
		Mapper<LongWritable, Text, Text, IntWritable>{
		private final static IntWritable one = new IntWritable(1);
		private Text word = new Text();
		public void map(LongWritable key, Text value, Context context)throws IOException, InterruptedException{
			String line = value.toString();
			StringTokenizer itr = new StringTokenizer(line);
			while(itr.hasMoreElements()){
				word.set(itr.nextToken());
				context.write(word, one);
			}
		}
	}
	public static  class wordcountReduce extends
		Reducer<Text, IntWritable, Text, IntWritable>{
		public void reduce(Text key, Iterable<IntWritable>values, Context context)throws IOException, InterruptedException{
			int sum = 0;
			for (IntWritable str : values){
				sum += str.get();
			}
			context.write(key, new IntWritable(sum));
		}
	}
	public static class MyMultiple extends MyMultipleOutputFormat{

		@Override
		protected String generateFileNameForKeyValue(WritableComparable key,
				Writable value, Configuration conf) {
			// TODO Auto-generated method stub
			return "other.txt";
		}
		
	}
	public static  void main(String args[])throws Exception{
		
		Configuration conf = new Configuration();

		Job job = new Job(conf, "wordcount");
		
		job.setJarByClass(WordCount_MulFileOut.class);
		
		job.setInputFormatClass(TextInputFormat.class);
		
		job.setOutputFormatClass(MyMultiple.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		
		job.setMapperClass(wordcountMapper.class);
		job.setReducerClass(wordcountReduce.class);
		job.setCombinerClass(wordcountReduce.class);
		
		FileInputFormat.setInputPaths(job, new Path(args[1]));
		FileOutputFormat.setOutputPath(job, new Path(args[2]));
		
		job.waitForCompletion(true);
	}
}