hadoop自定义文件输出格式

public class ImageOutputFormat extends TextOutputFormat<Text, DistanceVector> {
protected static class Writer extends RecordWriter<Text, DistanceVector> {
private LineRecordWriter<Text, Text> w;
public Writer(DataOutputStream dos, String codec) {
w = new LineRecordWriter<Text, Text>(dos, codec);
}
public Writer(DataOutputStream dos) {
w = new LineRecordWriter<Text, Text>(dos);
}
//  这里控制自己的输出style
public synchronized void write(Text key, DistanceVector value)
throws IOException {
StringBuffer sb = new StringBuffer();

System.out.println(value.getV2());
java.util.Iterator<Double> iter=value.getV1().iterator();
while(iter.hasNext())
{
sb.append(iter.next());new Text(key.toString());
sb.append(",");
}
sb.append(" distance:"+value.getV2());

w.write(new Text(key), new Text(sb.substring(0,
sb.length() - 1) + "\n"));
}

@Override
public void close(TaskAttemptContext context) throws IOException,
InterruptedException {
w.close(context);
}

;
}

@Override
public RecordWriter<Text, DistanceVector> getRecordWriter(
TaskAttemptContext job) throws IOException, InterruptedException {
Configuration conf = job.getConfiguration();
boolean isCompressed = getCompressOutput(job);
String keyValueSeparator = conf.get(
"mapred.textoutputformat.separator", "\t");//key value的分隔符自己可以设置
CompressionCodec codec = null;
String extension = "";
if (isCompressed) {
Class<? extends CompressionCodec> codecClass = getOutputCompressorClass(
job, GzipCodec.class);
codec = (CompressionCodec) ReflectionUtils.newInstance(codecClass,
conf);
extension = codec.getDefaultExtension();
}
Path file = getDefaultWorkFile(job, extension);
FileSystem fs = file.getFileSystem(conf);
if (!isCompressed) {
FSDataOutputStream fileOut = fs.create(file, false);
return new Writer(fileOut, keyValueSeparator);
} else {
FSDataOutputStream fileOut = fs.create(file, false);
return new Writer(new DataOutputStream(codec.createOutputStream(fileOut)), keyValueSeparator);
}
}

}