java - 具有 MultipleInputs 的 Hadoop 映射器的控制流程是什么？

Question

目标:实现Reduce Side Join

我的代码中目前有作业链(两个作业)。现在我想在减少端加入另一份工作。而且我必须接受多个输入:

Input #1:前一个 reducer 的输出。
Input #2:来自 HDFS 的新文件以实现连接。

看了一些关于如何使用的文章MultipleInputs.addInputhPath(job, Path, InputFormat.class, Mapper.class);

所以我明白我必须使用它两次，一次用于 Input #1一次是 Input #2 .

Question 1:那么如果我使用两个单独的映射器和一个单一的缩减器，哪个映射器将首先执行(或者它们将被并行执行)？如何在 reducer 端检查哪个 mapper 发出了 <key, value>对？

Question 2:如果我使用单个 mapper 和单个 reducer，控制流程会怎样？

Question 3:更多的 Hack，即不使用 MultipleInputs
使用DistributedCache是否可以(性能方面)？加载Input #2在 reducer 的setup()方法？并将前一个 reducer 的输出作为作业的唯一输入。

Note: Input #2文件很小。

Answer 1

答案 1:
Map如果插槽可用，则两个映射器的任务应该并行运行。单个插槽的存在可能会导致它们按顺序运行(可能会交错)，但这不是正常情况。如果 Mapper 序列有一些配置，我不知道这一点。

我再次怀疑 api可用于识别哪个 Mapper 发出了 <key, value> .准确的说，只是value的标识是必需的，因为相同的 key可能由不同的 map 发出。这通常是通过向输出值添加前缀标签并在 reducer 中解析这些标签来实现的。例如:

if(value.toString.startsWith("Input#1")){  //processing code }

看看这个blogpost ，它具有所有必需的提示和技巧。请注意，所有这些示例都使用旧的 mapred接口(interface)。但无论如何，逻辑都是一样的。

答案 2:
没有 MultipleInputs , 在 Map您必须使用可用的 Context 来识别传入对的文件名目的。例如:

public void map(LongWritable key, Text value, Context context)
    throws IOException, InterruptedException {
        String fileName = ((FileSplit) context.getInputSplit()).getPath().getName();   
        //............
}

然后只需在输出值前添加一个合适的标签，其余与 answer1 相同。

答案 3:
这很棘手。使用DistributedCache当要添加到缓存的文件很小时，加入可以提高性能。这可能是因为作业现在以较少数量的 Map 运行。任务。但它对大文件有不利影响。难题是要知道有多少字节被认为是 DistributedCache 的小字节。 .

既然您提到输入 #2 文件非常小，那么这应该是最适合您的解决方案。

注意:这篇文章的许多评论都是(有点)基于意见的。期待专家的意见。