如何在完成映射之前减少 Apache Spark 以减少内存使用？答案

【问题标题】：How do you get Apache Spark to reduce before finishing map to reduce memory usage?如何在完成映射之前减少 Apache Spark 以减少内存使用？
【发布时间】：2018-04-27 02:33:05
【问题描述】：

我正在使用 Apache Spark 执行 map-reduce 工作，但映射步骤会生成一个结构，该结构会占用大量内存。在将其他映射对象添加到内存之前，如何让它从内存中减少和删除映射？

我基本上是在做 myrdd.map(f).reduce(r)。但是， f 返回一个非常大的对象，所以我需要运行减速器，然后在太多堆积之前从内存中删除映射的对象。我能以某种方式做到这一点吗？

【问题讨论】：

这听起来有点像 XY 问题，但它确实会受益于一些代码，说明问题和语言标签。

标签： apache-spark

【解决方案1】：

类似于 MapReduce 中的组合器，在使用键/值对时，combineByKey() 接口可用于自定义组合器功能。像reduceByKey() 这样的方法默认使用它们自己的组合器在每个分区中本地组合数据，对于给定的键

类似于aggregate()（与单元素RDD一起使用），combineByKey()允许用户返回与输入RDD的元素类型相比不同的RDD元素类型。

【讨论】：

【解决方案2】：

trait SmallThing

trait BigThing

val mapFunction: SmallThing => BigThing = ???
val reduceFunction: (BigThing, BigThing) => BigThing = ???

val rdd: RDD[SmallThing] = ???

//initial implementation:
val result1: BigThing = rdd.map(mapFunction).reduce(reduceFunction)

//equivalent implementation:
val emptyBigThing: BigThing = ???
val result2: BigThing = rdd.aggregate(emptyBigThing)(seqOp = (agg, small) => reduceFunction(agg, mapFunction(small)), combOp = reduceFunction)

【讨论】：