apache-spark - Spark 可扩展性 : what am I doing wrong?

Question

我正在用 spark 处理数据，它可以处理一天的数据(40G)，但失败了 OOM 一周的数据:

import pyspark
import datetime
import operator
sc = pyspark.SparkContext()
sqc = pyspark.sql.SQLContext(sc)
sc.union([sqc.parquetFile(hour.strftime('.....'))
          .map(lambda row:(row.id, row.foo))
          for hour in myrange(beg,end,datetime.timedelta(0,3600))]) \
  .reduceByKey(operator.add).saveAsTextFile("myoutput")

不同ID的数量小于10k。
每个 ID 都是一个较小的 int .
作业失败是因为太多执行程序因 OOM 而失败。
当作业成功(在小输入上)时， "myoutput"大约是 100k。

我究竟做错了什么？

我尝试更换 saveAsTextFile与 collect (因为我实际上想在保存之前在 python 中进行一些切片和切块)，行为没有区别，同样的失败。这是意料之中的吗？

我以前有reduce(lambda x,y: x.union(y), [sqc.parquetFile(...)...])而不是 sc.union - 哪个更好？它有什么区别吗？

集群有 25 具有 的节点825GB RAM 和 224 其中的核心。

调用是 spark-submit --master yarn --num-executors 50 --executor-memory 5G .

单个 RDD 有大约 140 列并覆盖一小时的数据，因此一周是 168(=7*24) 个 RDD 的并集。

Answer 1

原来问题不在于 Spark ，而在于 yarn 。
解决方案是运行 spark

spark-submit --conf spark.yarn.executor.memoryOverhead=1000

(或修改 yarn 配置)。