scala - Spark RDD 在每个分区内具有共享指针(以及魔数(Magic Number) 200??)

Question

我试图保留一个 spark RDD，其中每个分区的元素都共享对单个大对象的访问。但是，这个对象似乎多次存储在内存中。将我的问题简化为只有 200 个元素的单个分区的玩具箱:

val nElements = 200
class Elem(val s:Array[Int])

val rdd = sc.parallelize(Seq(1)).mapPartitions( _ => {
    val sharedArray = Array.ofDim[Int](10000000) // Should require ~40MB
    (1 to nElements).toIterator.map(i => new Elem(sharedArray))
}).cache()

rdd.count() //force computation    

这会消耗预期的内存量，如日志中所示:

storage.MemoryStore: Block rdd_1_0 stored as values in memory (estimated size 38.2 MB, free 5.7 GB)

然而，200 是元素的最大数量。设置 nElements=201产量:

storage.MemoryStore: Block rdd_1_0 stored as values in memory (estimated size 76.7 MB, free 5.7 GB)

这是什么原因造成的？这个神奇的数字 200 来自哪里，我该如何增加它？

编辑澄清 :

向该函数添加一个 println 表明它确实只被调用了一次。此外，运行:

rdd.map(_.s.hashCode).min == rdd.map(_.s.hashCode).max  // returns true

..揭示所有 10000000 个元素确实指向同一个对象，因此数据结构基本上表现正确。当 nExamples 大得多(例如 20000)时就会出现问题，因此它无法持续存在。

storage.MemoryStore: Not enough space to cache rdd_1_0 in memory! (computed 6.1 GB so far)

当我设置 nExamples=500它成功地将 rdd 保留在内存中，说估计大小为 1907.4 MB，但我可以看到我的内存使用量的实际增加远小于这个。

Answer 1

对于将来遇到此问题的任何人，我最终想出了一个 super 黑客的解决方案(尽管我仍然很高兴听到更好的解决方案)。我没有使用 rdd.cache()，而是定义:

def cached[T: ClassTag](rdd:RDD[T]) = {
    rdd.mapPartitions(p => 
        Iterator(p.toSeq)
    ).cache().mapPartitions(p =>
        p.next().toIterator
    )
}

以便 cached(rdd)返回从“缓存”列表生成的 RDD