两个相等的组合键不会到达同一个减速器

我正在使用 MapReduce 框架用 Java 制作 Hadoop 应用程序。

我仅使用文本键和值进行输入和输出。在减少最终输出之前，我使用组合器进行额外的计算步骤。

但我有一个问题，钥匙没有进入同一个减速器。 我在组合器中创建并添加键/值对，如下所示：

public static class Step4Combiner extends Reducer<Text,Text,Text,Text> {
    private static Text key0 = new Text();
    private static Text key1 = new Text();

        public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
                key0.set("KeyOne");
                key1.set("KeyTwo");
                context.write(key0, new Text("some value"));
                context.write(key1, new Text("some other value"));
        }

}   

public static class Step4Reducer extends Reducer<Text,Text,Text,Text> {

            public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
                System.out.print("Key:" + key.toString() + " Value: ");
                String theOutput = "";
                for (Text val : values) {
                    System.out.print("," + val);
                }
                System.out.print("\n");

                context.write(key, new Text(theOutput));
            }

}

主要是我创建了这样的工作：

Configuration conf = new Configuration();
String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();

Job job4 = new Job(conf, "Step 4");
job4.setJarByClass(Step4.class);

job4.setMapperClass(Step4.Step4Mapper.class);
job4.setCombinerClass(Step4.Step4Combiner.class);
job4.setReducerClass(Step4.Step4Reducer.class);

job4.setInputFormatClass(TextInputFormat.class);
job4.setOutputKeyClass(Text.class);
job4.setOutputValueClass(Text.class);

FileInputFormat.addInputPath(job4, new Path(outputPath));
FileOutputFormat.setOutputPath(job4, new Path(finalOutputPath));            

System.exit(job4.waitForCompletion(true) ? 0 : 1);

从reducer打印的stdout中的输出是这样的：

Key:KeyOne Value: ,some value
Key:KeyTwo Value: ,some other value
Key:KeyOne Value: ,some value
Key:KeyTwo Value: ,some other value
Key:KeyOne Value: ,some value
Key:KeyTwo Value: ,some other value

这是没有意义的，因为键是相同的，因此它应该是 2 个减速器，其中 Iterable 中有 3 个相同的值

希望你能帮我弄清楚这个问题:)

这很可能是因为您的组合器在映射和减少阶段运行（一个鲜为人知的“功能”）。

基本上，您正在修改组合器中的键，当映射输出在减速器中合并在一起时，该键可能会运行，也可能不会运行。组合器运行后（reduce 端），键将通过分组比较器馈送，以确定将 Iterable 传递给reduce 方法的值（我在这里绕过了 reduce 阶段的流方面 - 不支持 iterable）通过值的集合或列表，如果分组比较器确定当前键和最后一个键相同，则对 iterator().next() 的更多调用将返回 true）

您可以尝试通过检查上下文（有一个Context.getTaskAttempt().isMap()方法，但我记得这也有问题，甚至可能在某处有关于此的 JIRA 票证）。

最重要的是，不要修改组合器中的密钥，除非您能找到绕过此行为的方法if组合器正在减少侧运行。

EDIT因此，在调查@Amar 的评论时，我整理了一些代码（粘贴链接 http://pastebin.com/PH5TsN6x），它添加了一些详细的比较器、组合器、缩减器等。如果您运行单个映射作业，那么在缩减阶段，不会运行组合器，并且映射输出不会再次排序，因为它已经被假定为已排序。

假设它是在发送到组合器类之前进行排序的，并且假设键将保持不变 - 因此仍然是排序的。请记住，组合器旨在组合给定键的值。

因此，使用单个映射和给定的组合器，reducer 会按 KeyOne、KeyTwo、KeyOne、KeyTwo、KeyOne 顺序查看键。分组比较器看到它们之间的转换，因此您会得到 6 次对reduce 函数的调用

如果您使用两个映射器，那么减速器知道它有两个已排序的段（每个映射一个），因此仍然需要在缩减之前对它们进行排序 - 但由于段的数量低于阈值，因此排序是作为内联流排序（再次假设段已排序）。使用两个映射器仍然会得到错误的输出（reduce 阶段输出 10 条记录）。

再次强调，不要修改组合器中的密钥，这不是组合器的用途。