hadoop - mapreduce 中的默认排序是使用 WritableComparable 类中定义的 Comparator 还是 comapreTo() 方法？

Question

在输出从 mapper 传递到 reducer 之前，mapreduce 中如何进行排序。如果我的映射器输出键是 IntWritable 类型，它是否使用 IntWritable 类中定义的比较器或类中的 compareTo 方法，如果是，如何进行调用。如果不是如何执行排序，如何进行调用？

Answer 1

Map 作业输出首先被收集，然后发送到 Partitioner，负责确定将数据发送到哪个 Reducer(尽管它尚未通过 reduce() 调用分组)。默认的 Partitioner 使用 Key 的 hashCode() 方法和 Reducer 数量的模数来做到这一点。

之后，将调用 Comparator 对 Map 输出执行排序。流程看起来像这样:

收集器 --> 分区器 --> 溢出 --> 比较器 --> 本地磁盘(HDFS)<-- MapOutputServlet

然后每个 Reducer 将从分区器分配给它的映射器复制数据，并将其传递给 Grouper，Grouper 将确定如何为单个 Reducer 函数调用对记录进行分组:

MapOutputServlet --> Copy to Local Disk (HDFS) --> Group --> Reduce

在函数调用之前，记录还将经过排序阶段以确定它们以何种顺序到达 reducer。 Sorter (WritableComparator()) 将调用 Key 的 compareTo() (WritableComparable() 接口(interface))方法。

为了给您一个更好的主意，下面是您将如何为自定义组合键实现基本的 compareTo()、grouper 和 sorter:

public class CompositeKey implements WritableComparable<CompositeKey> {
    IntWritable primaryField = new IntWritable();
    IntWritable secondaryField = new IntWritable();

    public CompositeKey(IntWritable p, IntWritable s) {
        this.primaryField.set(p);
        this.secondaryField = s;
    }

    public void write(DataOutput out) throws IOException {
        this.primaryField.write(out);
        this.secondaryField.write(out);
    }

    public void readFields(DataInput in) throws IOException {
        this.primaryField.readFields(in);
        this.secondaryField.readFields(in);
    }

    // Called by the partitionner to group map outputs to same reducer instance
    // If the hash source is simple (primary type or so), a simple call to their hashCode() method is good enough
    public int hashCode() {
        return this.primaryField.hashCode();
    }

    @Override
    public int compareTo(CompositeKey other) {
        if (this.getPrimaryField().equals(other.getPrimaryField())) {
            return this.getSecondaryField().compareTo(other.getSecondaryField());
        } else {
            return this.getPrimaryField().compareTo(other.getPrimaryField());
        }
    }
}

public class CompositeGroupingComparator extends WritableComparator {
    public CompositeGroupingComparator() {
        super(CompositeKey.class, true);
    }

    @Override
    public int compare(WritableComparable a, WritableComparable b) {
        CompositeKey first = (CompositeKey) a;
        CompositeKey second = (CompositeKey) b;

        return first.getPrimaryField().compareTo(second.getPrimaryField());
    }
}

public class CompositeSortingComparator extends WritableComparator {
    public CompositeSortingComparator() {
        super (CompositeKey.class, true);
    }

    @Override
    public int compare (WritableComparable a, WritableComparable b){
        CompositeKey first = (CompositeKey) a;
        CompositeKey second = (CompositeKey) b;

        return first.compareTo(second);
    }
}