Spark groupBy vs repartition 加 mapPartitions

Question

我的数据集大约有 2000 万行，需要大约 8 GB 的 RAM。我正在使用 2 个执行程序运行我的工作，每个执行程序 10 GB RAM，每个执行程序 2 个内核。由于进一步的转换，数据应该一次性缓存。

我需要根据 4 个字段减少重复项（选择任何重复项）。两个选项：使用 groupBy 以及使用 repartition 和 mapPartitions。第二种方法允许您指定分区数，因此在某些情况下可以执行得更快，对吗？

能否请您解释一下哪个选项的性能更好？这两个选项的 RAM 消耗是否相同？

使用groupBy

dataSet
    .groupBy(col1, col2, col3, col4)
    .agg(
        last(col5),
        ...
        last(col17)
    );

使用repartition和mapPartitions

dataSet.sqlContext().createDataFrame(
    dataSet
        .repartition(parallelism, seq(asList(col1, col2, col3, col4)))
        .toJavaRDD()
        .mapPartitions(DatasetOps::reduce),
    SCHEMA
);

private static Iterator<Row> reduce(Iterator<Row> itr) {
    Comparator<Row> comparator = (row1, row2) -> Comparator
        .comparing((Row r) -> r.getAs(name(col1)))
        .thenComparing((Row r) -> r.getAs(name(col2)))
        .thenComparingInt((Row r) -> r.getAs(name(col3)))
        .thenComparingInt((Row r) -> r.getAs(name(col4)))
        .compare(row1, row2);

    List<Row> list = StreamSupport
        .stream(Spliterators.spliteratorUnknownSize(itr, Spliterator.ORDERED), false)
        .collect(collectingAndThen(toCollection(() -> new TreeSet<>(comparator)), ArrayList::new));

    return list.iterator();
}

Answer 1

The second approach allows you to specify num of partitions, and could perform faster because of this in some cases, right?

不是真的。这两种方法都允许您指定分区数 - 在第一种情况下通过

spark.conf.set("spark.sql.shuffle.partitions", parallelism)

然而，如果重复很常见，第二种方法本质上效率较低，因为它先洗牌，然后减少，跳过 map-side 减少（换句话说，它是另一种 group-by-key）。如果重复很少见，这不会有太大区别。

旁注 Dataset 已经提供 dropDuplicates variants, which take a set of columns, and first / last is not particular meaningful here (see discussion in ).

Spark groupBy vs repartition 加 mapPartitions

Spark groupBy vs repartition plus mapPartitions

apache-spark

apache-spark-sql

apache-spark-dataset