hadoop编程小技巧（7）---自定义输出文件格式以及输出到不同目录

时间：2023-02-03 20:57:06

代码测试环境：Hadoop2.4

应用场景：当需要定制输出数据格式时可以采用此技巧，包括定制输出数据的展现形式，输出路径，输出文件名称等。

Hadoop内置的输出文件格式有：

1）FileOutputFormat<K,V>  常用的父类；

2）TextOutputFormat<K,V> 默认输出字符串输出格式；

3）SequenceFileOutputFormat<K,V> 序列化文件输出；

4）MultipleOutputs<K,V> 可以把输出数据输送到不同的目录；

5） NullOutputFormat<K,V> 把输出输出到/dev/null中，即不输出任何数据，这个应用场景是在MR中进行了逻辑处理，同时输出文件已经在MR中进行了输出，而不需要在输出的情况；

6）LazyOutputFormat<K,V> 只有在调用write方法是才会产生文件，这样的话，如果没有调用write就不会产生空文件；

步骤：

类似输入数据格式，自定义输出数据格式同样可以参考下面的步骤

1） 定义一个继承自OutputFormat的类，不过一般继承FileOutputFormat即可；

2）实现其getRecordWriter方法，返回一个RecordWriter类型；

3）自定义一个继承RecordWriter的类，定义其write方法，针对每个<key,Value>写入文件数据；


实例1（修改文件默认的输出文件名以及默认的key和value的分隔符）：

输入数据：

hadoop编程小技巧（7）---自定义输出文件格式以及输出到不同目录

自定义CustomFileOutputFormat(把默认文件名前缀替换掉):

package fz.outputformat;
import java.io.IOException;

import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class CustomOutputFormat extends FileOutputFormat<LongWritable, Text> {

private String prefix = "custom_";
@Override
public RecordWriter<LongWritable, Text> getRecordWriter(TaskAttemptContext job)
throws IOException, InterruptedException {
// 新建一个可写入的文件
Path outputDir = FileOutputFormat.getOutputPath(job);
//System.out.println("outputDir.getName():"+outputDir.getName()+",otuputDir.toString():"+outputDir.toString());
String subfix = job.getTaskAttemptID().getTaskID().toString();
Path path = new Path(outputDir.toString()+"/"+prefix+subfix.substring(subfix.length()-5, subfix.length()));
FSDataOutputStream fileOut = path.getFileSystem(job.getConfiguration()).create(path);
return new CustomRecordWriter(fileOut);
}

}
自定义CustomWriter（指定key，value分隔符）： 

package fz.outputformat;import java.io.IOException;import java.io.PrintWriter;import org.apache.hadoop.fs.FSDataOutputStream;import org.apache.hadoop.io.LongWritable;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapreduce.RecordWriter;import org.apache.hadoop.mapreduce.TaskAttemptContext;public class CustomRecordWriter extends RecordWriter<LongWritable, Text> {private PrintWriter out;private String separator =",";public CustomRecordWriter(FSDataOutputStream fileOut) {out = new PrintWriter(fileOut);}@Overridepublic void write(LongWritable key, Text value) throws IOException,InterruptedException {out.println(key.get()+separator+value.toString());}@Overridepublic void close(TaskAttemptContext context) throws IOException,InterruptedException {out.close();}}

调用主类： 

package fz.outputformat;import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.conf.Configured;import org.apache.hadoop.fs.Path;import org.apache.hadoop.io.LongWritable;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapreduce.Job;import org.apache.hadoop.mapreduce.Mapper;import org.apache.hadoop.mapreduce.Reducer;import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;import org.apache.hadoop.util.Tool;import org.apache.hadoop.util.ToolRunner;public class FileOutputFormatDriver extends Configured implements Tool{/** * @param args * @throws Exception  */public static void main(String[] args) throws Exception {// TODO Auto-generated method stubToolRunner.run(new Configuration(), new FileOutputFormatDriver(),args);}@Overridepublic int run(String[] arg0) throws Exception {if(arg0.length!=3){System.err.println("Usage:\nfz.outputformat.FileOutputFormatDriver <in> <out> <numReducer>");return -1;}Configuration conf = getConf();Path in = new Path(arg0[0]);Path out= new Path(arg0[1]);boolean delete=out.getFileSystem(conf).delete(out, true);System.out.println("deleted "+out+"?"+delete);Job job = Job.getInstance(conf,"fileouttputformat test job");job.setJarByClass(getClass());job.setInputFormatClass(TextInputFormat.class);job.setOutputFormatClass(CustomOutputFormat.class);job.setMapperClass(Mapper.class);job.setMapOutputKeyClass(LongWritable.class);job.setMapOutputValueClass(Text.class);job.setOutputKeyClass(LongWritable.class);job.setOutputValueClass(Text.class);job.setNumReduceTasks(Integer.parseInt(arg0[2]));job.setReducerClass(Reducer.class);FileInputFormat.setInputPaths(job, in);FileOutputFormat.setOutputPath(job, out);return job.waitForCompletion(true)?0:-1;}}

查看输出：

hadoop编程小技巧（7）---自定义输出文件格式以及输出到不同目录hadoop编程小技巧（7）---自定义输出文件格式以及输出到不同目录

从输出结果可以看到输出格式以及文件名确实按照预想输出了。


实例2（根据key和value值输出数据到不同目录）：
自定义主类（主类其实就是修改了输出的方式而已）：

package fz.multipleoutputformat;import org.apache.hadoop.conf.Configuration;import org.apache.hadoop.conf.Configured;import org.apache.hadoop.fs.Path;import org.apache.hadoop.io.LongWritable;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapreduce.Job;import org.apache.hadoop.mapreduce.Mapper;import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;import org.apache.hadoop.util.Tool;import org.apache.hadoop.util.ToolRunner;public class FileOutputFormatDriver extends Configured implements Tool{/** * @param args * @throws Exception  */public static void main(String[] args) throws Exception {// TODO Auto-generated method stubToolRunner.run(new Configuration(), new FileOutputFormatDriver(),args);}@Overridepublic int run(String[] arg0) throws Exception {if(arg0.length!=3){System.err.println("Usage:\nfz.multipleoutputformat.FileOutputFormatDriver <in> <out> <numReducer>");return -1;}Configuration conf = getConf();Path in = new Path(arg0[0]);Path out= new Path(arg0[1]);boolean delete=out.getFileSystem(conf).delete(out, true);System.out.println("deleted "+out+"?"+delete);Job job = Job.getInstance(conf,"fileouttputformat test job");job.setJarByClass(getClass());job.setInputFormatClass(TextInputFormat.class);//job.setOutputFormatClass(CustomOutputFormat.class);MultipleOutputs.addNamedOutput(job, "ignore", TextOutputFormat.class,LongWritable.class, Text.class);MultipleOutputs.addNamedOutput(job, "other", TextOutputFormat.class,LongWritable.class, Text.class);job.setMapperClass(Mapper.class);job.setMapOutputKeyClass(LongWritable.class);job.setMapOutputValueClass(Text.class);job.setOutputKeyClass(LongWritable.class);job.setOutputValueClass(Text.class);job.setNumReduceTasks(Integer.parseInt(arg0[2]));job.setReducerClass(MultipleReducer.class);FileInputFormat.setInputPaths(job, in);FileOutputFormat.setOutputPath(job, out);return job.waitForCompletion(true)?0:-1;}}
自定义reducer（因为要根据key和value的值输出数据到不同目录，所以需要自定义逻辑） 

package fz.multipleoutputformat;import java.io.IOException;import org.apache.hadoop.io.LongWritable;import org.apache.hadoop.io.Text;import org.apache.hadoop.mapreduce.Reducer;import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;public class MultipleReducer extendsReducer<LongWritable, Text, LongWritable, Text> {private MultipleOutputs<LongWritable,Text> out;@Overridepublic void setup(Context cxt){out = new MultipleOutputs<LongWritable,Text>(cxt);}@Overridepublic void reduce(LongWritable key ,Iterable<Text> value,Context cxt)throws IOException,InterruptedException{for(Text v:value){if(v.toString().startsWith("ignore")){//System.out.println("ignore--------------------value:"+v);out.write("ignore", key, v, "ign");}else{//System.out.println("other---------------------value:"+v);out.write("other", key, v, "oth");}}}@Overridepublic void cleanup(Context cxt)throws IOException,InterruptedException{out.close();}}

查看输出：

hadoop编程小技巧（7）---自定义输出文件格式以及输出到不同目录

可以看到输出的数据确实根据value的不同值被写入了不同的文件目录中，但是这里同样可以看到有默认的文件生成，同时注意到这个文件的大小是0，这个暂时还没解决。


总结：自定义输出格式，可以定制一些特殊需求，不过一般使用Hadoop内置的输出格式即可，这点来说其应用意义不是很大。不过使用Hadoop内置的MultipleOutputs可以根据数据的不同特性输出到不同的目录，还是很有实际意义的。


分享，成长，快乐

转载请注明blog地址：http://blog.csdn.net/fansy1990


标签：

相关文章