Apache Spark：在Java中有效地使用mapPartitions

Question

在目前早期发布的名为高性能Spark的教科书中，Spark的开发人员注意到：

为了使Spark能够灵活地将一些记录溢出到磁盘上，在MapPartitions中表示函数是很重要的，这样函数就不会强制将整个分区加载到内存中（例如隐式转换为列表）。迭代器有许多方法，我们可以在上面编写函数样式转换，或者您可以构造自己的自定义迭代器。当一个转换直接获取并返回一个迭代器而不强制它通过另一个集合时，我们称之为迭代器到迭代器转换。

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

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的这8行代码可以写成带有Map或FlatMap的一行代码，但我有意使用MapPartitions来利用这样一个事实，即它在每个分区上操作，而不是在RDD中的每个元素上操作。

有什么想法吗？

Answer 1

防止强制“物化”整个分区的一种方法是将迭代器转换为流，然后使用Stream的函数API（例如map函数）。

如何将迭代器转换为流？建议了一些将迭代器转换为流的好方法，因此采用这里建议的选项之一，我们可以得到以下结果：

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评论说，我引用，“在列表上使用流并没有提高效率”。我不知道为什么会这样，我也不知道总的来说这是否是真的，但值得一提。