java - Apache Spark : Effectively using mapPartitions in Java

Question

在当前名为 High Performance Spark 的早期发行教科书中，Spark 的开发人员指出:

To allow Spark the flexibility to spill some records to disk, it is important to represent your functions inside of mapPartitions in such a way that your functions don’t force loading the entire partition in-memory (e.g. implicitly converting to a list). Iterators have many methods we can write functional style transformations on, or you can construct your own custom iterator. When a transformation directly takes and returns an iterator without forcing it through another collection, we call these iterator-to-iterator transformations.

但是，教科书缺少使用 mapPartitions 的好例子或该方法的类似变体。在线上很少有好的代码示例——其中大部分是 Scala。例如，我们看到这个 Scala 代码使用 mapPartitions由 zero323 在 How to add columns into org.apache.spark.sql.Row inside of mapPartitions 上撰写.

def transformRows(iter: Iterator[Row]): Iterator[Row] = iter.map(transformRow)
sqlContext.createDataFrame(df.rdd.mapPartitions(transformRows), newSchema).show

不幸的是，Java 没有提供像iter.map(...) 这样好的东西。对于迭代器。所以它回避了一个问题，如何通过 mapPartitions 有效地使用迭代器到迭代器的转换？没有完全溢出 RDD以列表形式写入磁盘？

JavaRDD<OutObj> collection = prevCollection.mapPartitions((Iterator<InObj> iter) -> {
    ArrayList<OutObj> out = new ArrayList<>();
    while(iter.hasNext()) {
        InObj current = iter.next();
        out.add(someChange(current));
    }
    return out.iterator();
});

这似乎是使用 mapPartitions 的一般语法在 Java 示例中，但我看不出这是最有效的，假设您有一个 JavaRDD拥有数以万计的记录(甚至更多……因为 Spark 是用于大数据的)。您最终会得到迭代器中所有对象的列表，只是为了将其变回迭代器(这意味着某种 map 函数在这里会更有效率)。

注意:虽然这8行代码使用mapPartitions可以用 map 写成 1 行或 flatMap ，我有意使用 mapPartitions利用它对每个分区而不是 RDD 中的每个元素进行操作这一事实.

有什么想法吗？

Answer 1

防止强制“实现”整个分区的一种方法是将 Iterator 转换为 Stream，然后使用 Stream 的功能 API(例如 map 函数)。

How to convert an iterator to a stream?提出了一些将 Iterator 转换为 Stream 的好方法，因此采用那里建议的选项之一我们可以得到:

rdd.mapPartitions((Iterator<InObj> iter) -> {
    Iterable<InObj> iterable = () -> iter;
    return StreamSupport.stream(iterable.spliterator(), false)
            .map(s -> transformRow(s)) // or whatever transformation
            .iterator();
});

这应该是一个“Itrator-to-Iterator”转换，因为所有使用的中间 API(Iterable、Stream)都是延迟求值的。

编辑:我自己还没有测试过，但 OP 评论说，我引用，“在列表上使用 Stream 不会提高效率”。我不知道为什么会这样，我不知道一般情况下是否如此，但值得一提。