在每个分区内用共享指针触发RDDs（以及神奇的数字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    

这是什么原因造成的？这个神奇的数字200从何而来，我怎样才能增加它？

编辑以获得澄清：

向函数中添加一个println会显示它实际上只被调用一次。此外，运行：

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

Answer 1

对于将来遇到这个问题的人来说，我最终想出了一个超级黑客的解决方案（尽管我仍然很乐意听到更好的解决方案）。我不使用rdd.cache()，而是定义：

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

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