我遇到了Spark应用程序的问题。这是我的代码的简化版本:
def main(args:Array [String]){ //初始化spark上下文 val sc = new SparkContext() val …
好的,经过大量测试后,我终于解决了这个问题。实际上有两个问题:
的 我低估了输入数据的大小: 强> 星火公司 cache 要么 persist 如果RDD太大而无法完全存储在总内存的60%中,则函数效率不高,我知道但是我认为我的输入数据不是那么大但实际上我的RDD是80GB。但是我的60%的内存(160GB)仍然超过80GB,所以发生了什么?在问题n°2中回答...
cache
persist
的 我的分区太大了: 强> 在我的代码中,我的RDD的分区数设置为100,所以我有100个分区,每个1.6GB。问题是我的数据由每个数十个Megs的字符串组成,因此我的分区不满,10GB的已用内存实际上只包含7或8GB的实际数据。
为了解决这些问题,我不得不使用 persist(StorageLevel.MEMORY_SER) 这会增加计算时间,但会大大减少内存使用量 根据这个基准 )并将分区号设置为1000(根据Spark文档推荐分区~128MB)
persist(StorageLevel.MEMORY_SER)
你试过传递你的吗? modes.map 结果成为List构造函数(即 List(modes.map{ /*...*/}) )?有时(我不确定什么时候)Scala集合懒惰评估映射,所以如果直到spark删除缓存之后才对它们进行评估,那么它将不得不重新计算。
modes.map
List(modes.map{ /*...*/})