Hadoop0.20+ custom MultipleOutputFormat

Hadoop0.20.2中无法使用MultipleOutputFormat，多文件输出这个方法。尽管0.19.2中的方法老的方法org.apache.hadoop.mapred.lib.MultipleOutputFormat还是可以继续在0.20.2中使用，但是org.apache.hadoop.mapred下的方法都是标记为“已过时”，在hadoop下个版本中可能就不能使用了。hadoop 0.20.2中是推荐使用Configuration替换JobConf，而这个老的方法org.apache.hadoop.mapred.lib.MultipleOutputFormat中还是使用的JobConf，就是说还没有新的可替换API。

此外hadoop 0.20.2还只是一个中间版本，并不是所有API都升级到最新了，没有提供的API只能自己写。

重写MultipleOutputFormat需要2个类:

LineRecordWriter

MultipleOutputFormat

PartitionByFilenameOutputFormat是实验中需要自定义的每个文件各自输出结果

LineRecordWriter：

1. package cn.xmu.dm;   
2.   
3. import java.io.DataOutputStream;   
4. import java.io.IOException;   
5. import java.io.UnsupportedEncodingException;   
6. import org.apache.hadoop.io.NullWritable;   
7. import org.apache.hadoop.io.Text;   
8. import org.apache.hadoop.mapreduce.RecordWriter;   
9. import org.apache.hadoop.mapreduce.TaskAttemptContext;   
10. import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;   
11.   
12. public class LineRecordWriter<K, V> extends RecordWriter<K, V> {   
13.     private static final String utf8 = "UTF-8";   
14.   
15.     protected DataOutputStream out;   
16.     private final byte[] keyValueSeparator;   
17.   
18.     public LineRecordWriter(DataOutputStream out, String keyValueSeparator) {   
19.         this.out = out;   
20.         try {   
21.             this.keyValueSeparator = keyValueSeparator.getBytes(utf8);   
22.         } catch (UnsupportedEncodingException uee) {   
23.             throw new IllegalArgumentException("can't find " + utf8   
24.                     + " encoding");   
25.         }   
26.     }   
27.   
28.     public LineRecordWriter(DataOutputStream out) {   
29.         this(out, "/t");   
30.     }   
31.   
32.     private void writeObject(Object o) throws IOException {   
33.         if (o instanceof Text) {   
34.             Text to = (Text) o;   
35.             out.write(to.getBytes(), 0, to.getLength());   
36.         } else {   
37.             out.write(o.toString().getBytes(utf8));   
38.         }   
39.     }   
40.   
41.     public synchronized void write(K key, V value) throws IOException {   
42.         boolean nullKey = key == null || key instanceof NullWritable;   
43.         boolean nullValue = value == null || value instanceof NullWritable;   
44.         if (nullKey && nullValue) {   
45.             return;   
46.         }   
47.         if (!nullKey) {   
48.             writeObject(key);   
49.         }   
50.         if (!(nullKey || nullValue)) {   
51.             out.write(keyValueSeparator);   
52.         }   
53.         if (!nullValue) {   
54.             writeObject(value);   
55.         }   
56.         out.write("\r\n".getBytes());   
57.     }   
58.   
59.     public synchronized void close(TaskAttemptContext context)   
60.             throws IOException {   
61.         out.close();   
62.     }   
63. }