Hadoop 作业使用相同的减速器输出到相同的文件

Question

我遇到了一个有趣的情况，现在正在寻找如何故意这样做。在我的本地单节点设置中，我从终端屏幕同时运行了 2 个作业。我的两个作业都使用相同的reducer，它们仅在映射函数（聚合键-分组依据）方面有所不同，两个作业的输出都写入了第一个作业的输出（尽管第二个作业确实创建了自己的文件夹，但它是空的）。我正在做的是提供跨各个级别的汇总聚合，这种行为对我来说很有吸引力，我可以在一个文件中使用来自两个不同级别的聚合输出（也完美排序）。

我的问题是如何在真正的 Hadoop 集群中实现相同的目标，我们有多个数据节点，即我以编程方式启动多个作业，所有作业都访问相同的输入文件，以不同方式映射数据，但使用相同的 reducer，输出为在一个文件中可用，而不是在 5 个不同的输出文件中。

请指教。

在我决定提问之前，我正在查看merge output files after reduce phase。

感谢和亲切的问候，

莫伊兹·艾哈迈德。

Answer 1

当不同的 Mapper 使用相同的输入文件，也就是相同的数据结构时，所有这些不同的 Mapper 的源代码可以放在单个 Mapper 实现的不同方法中，并使用上下文中的参数来决定哪个映射要调用的函数。从好的方面来说，您只需要启动一个 Map Reduce Job。例子是伪代码：

class ComplexMapper extends Mapper {

protected BitSet mappingBitmap = new BitSet();

protected void setup(Context context) ... {
{
    String params = context.getConfiguration().get("params");
    ---analyze params and set bits into the mappingBitmap
}

protected void mapA(Object key, Object value, Context context){
.....
context.write(keyA, value);
}


protected void mapB(Object key, Object value, Context context){
.....
context.write(keyA, value);
}


protected void mapB(Object key, Object value, Context context){
.....
context.write(keyB, value);
}

public void map(Object key, Object value, Context context) ..... {
   if (mappingBitmap.get(1)) {
       mapA(key, value, context);
   }
   if (mappingBitmap.get(2)) {
       mapB(key, value, context);
   }
   if (mappingBitmap.get(3)) {
       mapC(key, value, context);
   }
}

当然，它可以通过接口等更优雅地实现。

在作业设置中添加：

Configuration conf = new Configuration();
conf.set("params", "AB");

Job job = new Job(conf);

正如 Praveen Sripati 所提到的，拥有一个输出文件将迫使您只拥有一个 Reducer，这可能对性能不利。当您从 hdfs 下载 part** 文件时，您始终可以连接它们。示例：

hadoop fs -text /output_dir/part* > wholefile.txt

Answer 2

通常每个 reducer 任务在 HDFS 中生成一个单独的文件，以便 reduce 任务可以并行运行。如果要求从 reduce 任务中获得一个 o/p 文件，则将作业配置为具有一个 reducer 任务。可以使用默认为 1 的 mapred.reduce.tasks 属性配置减速器的数量。这种方法的缺点是只有一个减速器，这可能是完成工作的瓶颈。

另一种选择是使用其他一些输出格式，它允许多个减速器同时写入同一个接收器，如DBOuputFormat。作业处理完成后，可以将数据库中的结果导出到平面文件中。这种方法将使多个reduce任务并行运行。

另一个选项是合并 OP 中提到的 o/p 文件。因此，根据每种方法的优缺点以及要处理的数据量，可以选择其中一种方法。