自定义OutPutFormat

昨天学习了自定义InputFormat,今天又去看了看TextOutFormat.java的源码。一样比较简单。现在我们来实现自定义的OutPutFormat

先上代码：

public class MyOutputFormat<K, V> extends FileOutputFormat<K, V>{
//可以看到这里继承了FileOutputFormat，这个类一样可以自己改写。
	protected static class MyRecordWriter<K, V>
	extends RecordWriter<K, V> {
//这里实现了自己的RecordWriter，在读数据的时候就通过这个类来实现。
		private static final String utf8 = "UTF-8";
//为了简单，我将系统默认的"/t"分隔符变成了"----"，如果你想变更复杂也比较简单。比如你可以自定义个key,再按你需要的格式输出。
		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 MyRecordWriter(DataOutputStream out) {
	        this(out, colon); //调用下面的构造函数
	    }
		public MyRecordWriter(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);
			}
			
		}
	}
//这个将被自动调用来读数据并写往文件系统。
	@Override
	public RecordWriter<K, V> getRecordWriter(TaskAttemptContext job)
			throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		Configuration conf = job.getConfiguration();
		
		Path file = getDefaultWorkFile(job, "");
		FileSystem fs = file.getFileSystem(conf);
		FSDataOutputStream fileOut = fs.create(file, false);
	      return new MyRecordWriter<K, V>(fileOut);
	}
}

这个自定义输出也是最简单的。到了真正应用的时候就可以根据需要自己构造输出格式。

比如我们可以不用系统自带的 Text, IntWritable, LongWritable数据类型。自己写一个自定义数据类型，用自定义的格式输出。

测试的代码我就不上了，上次那个自定义InputFormat文章里已经贴出啦了。只需要多加上一句：

job.setOutputFormatClass(MyOutputFormat.class);