实现mapreduce多文件自定义输出
普通maprduce中通常是有map和reduce两个阶段，在不做设置的情况下，计算结果会以part-000*输出成多个文件，并且输出的文件数量和reduce数量一样，文件内容格式也不能随心所欲。这样不利于后续结果处理。
在hadoop中，reduce支持多个输出,输出的文件名也是可控的，就是继承MultipleTextOutputFormat类，重写generateFileNameForKey方法。如果只是想做到输出结果的文件名可控，实现自己的LogNameMultipleTextOutputFormat类，设置jobconf.setOutputFormat(LogNameMultipleTextOutputFormat.class);就可以了，但是这种方式只限于使用旧版本的hadoop api.如果想采用新版本的api接口或者自定义输出内容的格式等等更多的需求，那么就要自己动手重写一些hadoop api了。
首先需要构造一个自己的MultipleOutputFormat类实现FileOutputFormat类（注意是org.apache.hadoop.mapreduce.lib.output包的FileOutputFormat）
[java]
import java.io.DataOutputStream;
import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Writable;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.GzipCodec;
import org.apache.hadoop.mapreduce.OutputCommitter;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.ReflectionUtils;
/**
* This abstract class extends the FileOutputFormat, allowing to write the
* output data to different output files. There are three basic use cases for
* this class.
* Created on 2012-07-08
* @author zhoulongliu
* @param <K>
* @param <V>
*/
public abstract class MultipleOutputFormat<K extends WritableComparable<?>, V extends Writable> extends
FileOutputFormat<K, V> {
//接口类，需要在调用程序中实现generateFileNameForKeyValue来获取文件名
private MultiRecordWriter writer = null;
public RecordWriter<K, V> getRecordWriter(TaskAttemptContext job) throws IOException, InterruptedException {
if (writer == null) {
writer = new MultiRecordWriter(job, getTaskOutputPath(job));
}
return writer;
}
/**
* get task output path
* @param conf
* @return
* @throws IOException
*/
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来确定输出文件名（含扩展名） Generate the file output file name based
* on the given key and the leaf file name. The default behavior is that the
* file name does not depend on the key.
*
* @param key the key of the output data
* @param name the leaf file name
* @param conf the configure object
* @return generated file name
*/
protected abstract String generateFileNameForKeyValue(K key, V value, Configuration conf);
/**
* 实现记录写入器RecordWriter类
* （内部类）
* @author zhoulongliu
*
*/
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>>();
}
@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();
补充：软件开发 , Java ,
