gpt4 book ai didi

java - Apache Spark : Effectively using mapPartitions in Java

转载 作者:搜寻专家 更新时间:2023-10-30 21:28:06 24 4
gpt4 key购买 nike

在当前名为 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 中的每个元素进行操作这一事实.

有什么想法吗?

最佳答案

防止强制“实现”整个分区的一种方法是将 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(IterableStream)都是延迟求值的。

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

关于java - Apache Spark : Effectively using mapPartitions in Java,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/42539315/

24 4 0
Copyright 2021 - 2024 cfsdn All Rights Reserved 蜀ICP备2022000587号
广告合作:1813099741@qq.com 6ren.com