hadoop - 一组中的所有记录都调用一次 'reduce' 吗？

Question

我在 hadoop 中做了一个练习，用于对对象“IntPair”进行排序，它是 2 个整数的组合。这是输入文件:

2,9
3,8
2,6
3,2
...

'IntPair' 类是这样的:

static class IntPair implements WritableComparable<IntPair> {
    private int first;
    private int second;   
       ...
   public int compareTo(IntPair o) {
       return (this.first==o.first)?(this.second==o.second?0:(this.second>o.second?1:-1)):(this.first>o.first?1:-1);
    }
   public static int compare(int a, int b) {
   return (a==b)?0:((a>b)?1:-1);
   }
       ...  
}

在 Mapper 中，我使用了 inputFormat 和 outputKey/Value，并且只创建了每行 2 个整数的 IntPair 实例:

protected void map(LongWritable key, Text value, Context context)
        throws IOException, InterruptedException {
            String v[] = value.toString().split(",");
            IntPair k = new IntPair(Integer.parseInt(v[0]), Integer.parseInt(v[1]));
            context.write(k, NullWritable.get());

        }

我根据第一个整数对映射器结果进行分区，并根据第一个整数创建组比较器。只有排序比较器基于两个整数。

static class FirstPartitioner extends Partitioner<IntPair, NullWritable> {

    public int getPartition(IntPair key, NullWritable value, int numPartitions) {
            return Math.abs(key.getFirst()*127)%numPartitions;
        }
}
static class BothComparator extends WritableComparator {
    public int compare(WritableComparable w1, WritableComparable w2) {
            IntPair p1 = (IntPair)w1;
            IntPair p2 = (IntPair)w2;
            int cmp = IntPair.compare(p1.getFirst(), p2.getFirst());
            if(cmp != 0) {
                return cmp;
            }
            return -IntPair.compare(p1.getSecond(), p2.getSecond());//reverse sort
    }

}

static class FirstGroupComparator extends WritableComparator {
    public int compare(WritableComparable w1, WritableComparable w2) {
            IntPair p1 = (IntPair)w1;
            IntPair p2 = (IntPair)w2;
            return IntPair.compare(p1.getFirst(), p2.getFirst());
    }
}

在 Reducer 中，我只输出 IntPair 作为键，输出 NullWritable 作为值:

static class SSReducer extends Reducer<IntPair, NullWritable, IntPair, NullWritable> {
        protected void reduce(IntPair key, Iterable<NullWritable> values,
            Context context)throws IOException, InterruptedException {
            context.write(key, NullWritable.get());
        }
}

运行hadoop后得到如下结果:

   2,9
   3,8

之前，我认为reducer应该通过键(IntPair)对记录进行分组。由于每条记录代表一个不同的键，因此每条记录将调用方法“reduce”一次，在这种情况下结果应该是:

2,9
2,6
3,8
3,2

所以我认为差异的存在是因为组比较器，因为它只使用第一个整数进行比较。所以在 reducer 中，记录按第一个整数分组。在此示例中，这意味着 2 条记录中的每条记录都调用“reduce”一次，因此在不循环的情况下，它只生成每组的第一条记录。这样对吗？另外，我做了另一个实验，改变了 reducer 如下:

static class SSReducer extends Reducer<IntPair, NullWritable, IntPair, NullWritable> {
     protected void reduce(IntPair key, Iterable<NullWritable> values,
                Context context)throws IOException, InterruptedException {
                        for(NullWritable n : values) //add looping
                   context.write(key, NullWritable.get());
            }
    }

然后它产生的结果中有 4 个项目。

如果我将 groupcomparator 更改为使用两个整数进行比较，它也会产生 4 个项目。因此，reducer 实际上使用 groupcomparator 对键进行分组，这意味着即使键不同，一个组中的每个记录都会调用一次“reduce”。

Answer 1

是的，即使 key 不同，一组中的每个记录都会调用“reduce”一次。实际上，每个组调用一次 reduce 方法，组中的第一个键为 'KEY'，组中的所有值构成 reduce 方法的值。

即使我们在 reduce 方法中只有一个键(第一个键)并且所有值都是可迭代的，您可以看到在迭代时我们将获得与可迭代中的值对应的键。

首先，我们使用两个键转到 groupcomparator，reduce 方法开始，然后从迭代器内部再次使用另外 2 个键调用组比较器。

这意味着 reducer 事先不知道它的可迭代值。它是在迭代可迭代值时确定的。

因此，如果我们不迭代值，我们只会看到组中的第一个键。如果我们迭代值，我们将获得所有键。