如何使用新的 Hadoop API 来使用 MultipleTextOutputFormat？

Question

我想编写多个输出文件。 如何使用 Job 而不是 JobConf 来执行此操作？

Answer 1

创建基于密钥的输出文件名的简单方法

 input data type

  //key        //value
 cupertino   apple
 sunnyvale   banana
 cupertino   pear

MultipleTextOutputFormat 类

static class KeyBasedMultipleTextOutputForma extends MultipleTextOutputFormat<Text, Text> {
    @Override
    protected String generateFileNameForKeyValue(Text key, Text value, String name) {
        return key.toString() + "/" + name;
    }
} 

作业配置

 job.setOutputFormat(KeyBasedMultipleTextOutputFormat.class);

运行此代码，您将在 HDFS 中看到以下文件，其中 /output 是作业输出目录：

 $ hadoop fs -ls /output
 /output/cupertino/part-00000
 /output/sunnyvale/part-00000

希望它有所帮助。

Answer 2

文档说使用 org.apache.hadoop.mapreduce.lib.output.MultipleOutputs 代替。

下面是使用 MultipleOutputs 的代码片段。不幸的是，我没有写它，也没有花太多时间来处理它......所以我不知道为什么事情会在哪里。我分享希望它有帮助。 :)

作业设置

job.setJobName("Job Name");
job.setJarByClass(ETLManager.class);
job.setMapOutputKeyClass(Text.class);
job.setOutputKeyClass(NullWritable.class);
job.setMapOutputValueClass(MyThing.class);
job.setMapperClass(MyThingMapper.class);
job.setReducerClass(MyThingReducer.class);
MultipleOutputs.addNamedOutput(job, Constants.MyThing_NAMED_OUTPUT, TextOutputFormat.class, NullWritable.class, Text.class);
job.setInputFormatClass(MyInputFormat.class);
FileInputFormat.addInputPath(job, new Path(conf.get("input")));
FileOutputFormat.setOutputPath(job, new Path(String.format("%s/%s", conf.get("output"), Constants.MyThing_NAMED_OUTPUT)));

减速器设置

public class MyThingReducer extends
    Reducer<Text, MyThing, NullWritable, NullWritable> {
    private MultipleOutputs     m_multipleOutputs;

     @Override
    public void setup(Context context) {
        m_multipleOutputs = new MultipleOutputs(context);
    }
    @Override
    public void cleanup(Context context) throws IOException,
            InterruptedException {
        if (m_multipleOutputs != null) {
            m_multipleOutputs.close();
        }
    }

    @Override
    public void reduce(Text key, Iterable<MyThing> values, Context context)throws IOException, InterruptedException {
        for (MyThing myThing : values) {
            m_multipleOutputs.write(Constants.MyThing_NAMED_OUTPUT, EMPTY_KEY, generateData(context, myThing), generateFileName(context, myThing));
            context.progress();
        }
    }
}

编辑：添加了到 MultipleOutputs 的链接。