MapReduce中控制输出文件命名 & 单个reducer写出多个输出文件——MutipleOutputs

75 阅读 0 评论 50 点赞

我是靠谱客的博主活泼菠萝，最近开发中收集的这篇文章主要介绍MapReduce中控制输出文件命名 & 单个reducer写出多个输出文件——MutipleOutputs，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

在MR job中，可以使用FileInputFormat和FileOutputFormat来对输入路径和输出路径进行设置。

在输出目录中，框架自己会自动对输出文件进行组织和命名：

一般情况下，Hadoop中每个Reducer对一个相同key的value做归并后，产生一个输出文件，并且文件以part-r-00000，part-r-00001的方式命名；

但是如果需要人为控制Reducer端输出文件名称（name-r-nnnnn命名，name是指定的名字，nnnnn是分区号），

或者一个Reducer输出多个文件，需要用到MR框架中的MutipleOutputs，用法如下。

1. 在Reducer类中，声明MutipleOutputs对象mos，指定泛型

private static class BitMapCalculationReducer extends Reducer<Text, Text, Text, Text> {
    //泛型类型是Reducer端输出的k-v类型，也就是参数列表后两个
    private MultipleOutputs<Text, Text> mos;
}

2. 在setup方法中，给mos对象配置上资源环境context

@Override
protected void setup(Context context) throws IOException, InterruptedException {
    mos = new MultipleOutputs<>(context);
}

3. 在reduce方法中，使用mos.write（name，key，value，String baseOutputPath）代替context.write（key，value）输出

第一个参数name是文件名，key和outputValue都是Text类型

第四个参数指定输出文件目录，命名格式为baseOutputPath-r-nnnnn

输入和输出文件目录都是HDFS路径

@Override
protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
    mos.write("daytask", key, outputValue, outputDir);
}

4. 在cleanup方法中关闭输出流mos

@Override
protected void cleanup(Context context) throws IOException, InterruptedException {
    try{
            mos.close();
        }catch (Exception e){
            logger.error(e.getMessage());
        }finally {
            mos = null;
        }
}

5. 在job类中，用addNamedOutput设置MutipleOutputs的输出文件类型

五个参数分别是：job、文件名、MR中输出文件类型（二进制）、输出key类型、输出value类型

@Override
public int run(String[] args) throws Exception {
    设置job(conf)
    MultipleOutputs.addNamedOutput(
        job, "daytask", SequenceFileOutputFormat.class, Text.class, Text.class);
    }
}