Hadoop reducer 接收错误数据答案

【问题标题】：Hadoop reducers receiving wrong dataHadoop reducer 接收错误数据
【发布时间】：2013-03-08 09:48:44
【问题描述】：

我有大量 JobControls 同时运行，它们都具有相同的 ControlledJobs 集。每个 JobControl 按日期范围处理一组不同的输入/输出文件，但它们都是类型。我观察到的问题是，reduce 步骤正在接收旨在由处理不同日期范围的 reducer 处理的数据。日期范围由 Job 设置，用于确定输入和输出，并从 reducer 内的上下文中读取。

如果我按顺序提交 JobControl，这会停止，但这不好。这是我应该用自定义分区器解决的问题吗？如果我不知道哪个减速器正在处理我当前的日期范围，我什至如何确定密钥的正确减速器？为什么实例化的 reducer 不会被锁定到它们的 JobControl？

我已经针对 Java 中的基本实现编写了所有 JobControls、Jobs、Maps 和 Reduces。

我正在使用带有纱线的 2.0.3-alpha。会不会有什么关系？

我必须小心地共享代码，但这里有一个经过消毒的映射器：

protected void map(LongWritable key, ProtobufWritable<Model> value, Context context) 
    throws IOException, InterruptedException {
  context.write(new Text(value.get().getSessionId()), 
                new ProtobufModelWritable(value.get()));
}

还有减速器：

protected void reduce(Text sessionId, Iterable<ProtobufModelWritable> models, Context context) 
     throws IOException, InterruptedException {
  Interval interval = getIntervalFromConfig(context);
  Model2 model2 = collapseModels(Iterables.transform(models, TO_MODEL));

  Preconditions.checkArgument(interval.contains(model2.getTimeStamp()), 
      "model2: " + model2 + " does not belong in " + interval);
}

private Interval getIntervalFromConfig(Context context) {
  String i = context.getConfiguration().get(INTERVAL_KEY);
  return Utils.interval(i);
}

【问题讨论】：

你能详细说明一下吗？您使用的是哪种类型的密钥，能否向我们展示您的映射器代码？
您好 Thomas，已添加代码。希望这会有所帮助。
首先，感谢代码可读性好 +1！但是，您设置整个作业的时间间隔。因此，如果您有多个 reducer，那么您在所有 reduce 任务中的间隔都是相同的。所以如果你想在 reducer 中使用不同的间隔，你必须用你的日期范围分割逻辑覆盖分区器。
问题是，每个间隔我都有一个单独的作业。 Job 和 Reduce 类是相同的，但它们为每个实例化创建了不同的配置。这意味着有多个相同类的减速器同时运行，但用于不同的作业。
我会采用@ThomasJungblut 建议的方法 - 覆盖分区器并根据 interval_key 进行拆分。在您的地图任务中，您应该从配置中读取所需间隔列表并发出每个记录N 次，每次为另一个间隔。使用这种方法，您只需要一个作业，reducers 将并行计算所需的结果。

标签： java hadoop mapreduce hadoop-partitioning

【解决方案1】：

作为参考，我用两件事解决了这个问题。最重要的问题是，当我为每个间隔创建单独的作业时，我给它们每个都赋予了相同的名称。通过将序列化间隔附加到作业名称，Hadoop 知道将映射结果发送到哪些 reducer。

此外，我开始为每个作业创建单独的配置对象，而不是复制初始配置。这可能是不必要的，但至少我知道我不会犯错并开始共享同一个配置对象。

【讨论】：