火花可扩展性：我做错了什么？答案

【问题标题】：spark scalability: what am I doing wrong?火花可扩展性：我做错了什么？
【发布时间】：2015-05-23 06:10:10
【问题描述】：

我正在使用 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。

我做错了什么？
我尝试用collect 替换saveAsTextFile（因为我实际上想在保存之前在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 的并集。

【问题讨论】：

我将从查看性能 UI 开始，并记住 union 不是典型的 union。如果你想要一个独特的联合，那么你需要在联合之后调用 distinct。
@JustinPihony：我知道union 意味着连接，而这正是我想要的。什么是“性能 UI”？
spark.apache.org/docs/1.2.0/monitoring.html
如果只有 25 个节点，为什么还要调用 --num-executors 50？
@MikelUrkia：因为我有 224 个核心。

标签： apache-spark bigdata pyspark scalability distributed-computing

【解决方案1】：

Spark 在扩展时经常遇到内存不足错误。在这些情况下，程序员应该进行微调。或者重新检查你的代码，确保你没有做太多的事情，比如收集驱动程序中的所有bigdata，这很可能超过memoryOverhead限制，不管你设置多大。

要了解正在发生的事情，您应该意识到yarn 决定因超出内存限制而终止容器。当容器超出 memoryOverhead 限制时，就会发生这种情况。

在调度程序中，您可以检查事件时间线以查看容器发生了什么。如果 Yarn 杀死了一个容器，它将显示为红色，当您将鼠标悬停/单击它时，您将看到如下消息：

容器因超出内存限制而被 YARN 杀死。使用了 16.9 GB 的 16 GB 物理内存。考虑提升 spark.yarn.executor.memoryOverhead。

所以在这种情况下，您要关注的是这些配置属性（值是 my 集群上的示例）：

# More executor memory overhead
spark.yarn.executor.memoryOverhead          4096

# More driver memory overhead
spark.yarn.driver.memoryOverhead            8192

# Max on my nodes
#spark.executor.cores                        8
#spark.executor.memory                       12G

# For the executors
spark.executor.cores                        6
spark.executor.memory                       8G

# For the driver
spark.driver.cores                          6
spark.driver.memory                         8G

首先要做的是增加memoryOverhead。

在驱动程序中还是在执行程序中？

当您从 UI 概述集群时，您可以单击尝试 ID 并检查 诊断信息，其中应该提到被杀死的容器的 ID。如果它与您的 AM Container 相同，则它是驱动程序，否则是执行程序。

这并没有解决问题，现在怎么办？

您必须微调您提供的内核数量和堆内存。您会看到pyspark 将在堆外内存中完成大部分工作，因此您不希望为堆提供太多空间，因为那样会被浪费。你不想给的太少，因为垃圾收集器会出现问题。回想一下，这些是 JVM。

如here 所述，一个worker 可以托管多个执行器，因此使用的核心数会影响每个执行器的内存量，因此减少#cores 可能会有所帮助。

我用memoryOverhead issue in Spark 和 Spark – Container exited with a non-zero exit code 143 写了更详细的内容，大部分我不会忘记！另一个我没有尝试过的选项是spark.default.parallelism 或/和spark.storage.memoryFraction，根据我的经验，这没有帮助。

您可以像提到的 sds 一样传递配置标志，或者像这样：

spark-submit --properties-file my_properties

其中“my_properties”类似于我上面列出的属性。

对于非数值，你可以这样做：

spark-submit --conf spark.executor.memory='4G'

【讨论】：

【解决方案2】：

原来问题不在于火花，而在于纱线。解决方案是使用

运行 spark

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

（或修改纱线配置）。

【讨论】：