处理 Hadoop mapreduce 中键值的不均匀分布答案

【问题标题】：Handle uneven distribution of values across keys in Hadoop mapreduce处理 Hadoop mapreduce 中键值的不均匀分布
【发布时间】：2013-07-25 23:49:29
【问题描述】：

我正在处理 hadoop 中的输入日志文件，其中密钥分布不均。这意味着减速器的值分布不均匀。例如 key1 有 1 个值，而 key2 有 1000 个值。

有什么方法可以对与同一个键关联的值进行负载平衡[我也不想修改我的键]

【问题讨论】：

你能从算法的角度描述你的工作吗 - 一旦你的键进入减速器，你想对它们做什么（例如，它是一个 sum / min / max / avg 计算还是类似的？ - 是否可以将此计算的一部分迁移到组合器以减少映射器和缩减器之间针对倾斜键的数据流？）

标签： java hadoop mapreduce partitioning hadoop-partitioning

【解决方案1】：

如果您知道哪些键将具有异常大量的值，则可以使用以下技巧。

您可以实现自定义Partitioner，这将确保您的每个倾斜键都进入单个分区，然后其他所有内容将通过它们的hashCode（这是默认的@987654323 @ 确实）。

您可以通过实现此接口来创建自定义Partitioner：

public interface Partitioner<K, V> extends JobConfigurable {
  int getPartition(K key, V value, int numPartitions);
}

然后您可以告诉 Hadoop 使用您的 Partitioner 与：

conf.setPartitionerClass(CustomPartitioner.class);

【讨论】：

非常感谢@charles。不幸的是，我不知道哪些所有的键都会有大量的值。同样在您的解决方案中，这种方法将导致特定的 reducer [接收 1000 个值的那个] 处理大量数据。我担心的原因是，因为对于属于特定键的每个值，我都会进行大量计算[您可以说某些键将具有 75000 个值，并且我会遍历 reducer 中的值并进行一些计算，每个计算需要 2 分钟]

【解决方案2】：

也许您可以在使用减速器之前使用组合器？这是相当投机的......

想法是将每组键划分为预设最大大小的分区，然后将这些分区的 k/v 对输出到 reducer。此代码假定您已在配置中的某处设置了该大小。

public static class myCombiner extends Reducer<Text, Text, Text, Text> {
    public void reduce(Text key, Iterable<Text> values, Context context)
            throws IOException, InterruptedException {

        List<Text> textList = new ArrayList<Text>();
        int part = 0;

        while (values.iterator().hasNext()) {
            if (textList.size() <= Integer.parseInt(context.getConfiguration().get("yourMaxSize"))) {
                textList.add(values.iterator().next());

            } else {
                for(Text t : textList) {
                    //essentially partitioning each key...
                    context.write(new Text(key.toString() + "_" + Integer.toString(part)), t);
                }
                textList.clear();
            }
            part += 1;
        }
        //output any stragglers ... 
        for(Text t : textList) {
            context.write(new Text(key.toString() + "_" + Integer.toString(part)), t);
        }

    }
}

【讨论】：