执行器心跳超时 Spark on DataProc答案

【问题标题】：Executor heartbeat timed out Spark on DataProc执行器心跳超时 Spark on DataProc
【发布时间】：2017-01-11 13:05:46
【问题描述】：

我正在尝试在 Google DataProc 集群上的 Spark (2.0.0) 中拟合 ml 模型。拟合模型时，我收到 Executor heartbeat timed out 错误。我该如何解决这个问题？

其他解决方案表明这可能是由于（其中一个）执行程序内存不足。我作为解决方案阅读：设置正确的设置、重新分区、缓存并获得更大的集群。我能做什么，最好不要设置更大的集群？（多/少分区？少缓存？调整设置？）

我的设置：

Google DataProc 集群上的 Spark 2.0.0： 1 个 Master 和 2 个 worker 都具有相同的规格：n1-highmem-8 -> 8 个 vCPU，52.0 GB 内存 - 500GB 磁盘

设置：

spark\:spark.executor.cores=1
distcp\:mapreduce.map.java.opts=-Xmx2457m
spark\:spark.driver.maxResultSize=1920m
mapred\:mapreduce.map.java.opts=-Xmx2457m
yarn\:yarn.nodemanager.resource.memory-mb=6144
mapred\:mapreduce.reduce.memory.mb=6144
spark\:spark.yarn.executor.memoryOverhead=384
mapred\:mapreduce.map.cpu.vcores=1
distcp\:mapreduce.reduce.memory.mb=6144
mapred\:yarn.app.mapreduce.am.resource.mb=6144
mapred\:mapreduce.reduce.java.opts=-Xmx4915m
yarn\:yarn.scheduler.maximum-allocation-mb=6144
dataproc\:dataproc.scheduler.max-concurrent-jobs=11
dataproc\:dataproc.heartbeat.master.frequency.sec=30
mapred\:mapreduce.reduce.cpu.vcores=2
distcp\:mapreduce.reduce.java.opts=-Xmx4915m
distcp\:mapreduce.map.memory.mb=3072
spark\:spark.driver.memory=3840m
mapred\:mapreduce.map.memory.mb=3072
yarn\:yarn.scheduler.minimum-allocation-mb=512
mapred\:yarn.app.mapreduce.am.resource.cpu-vcores=2
spark\:spark.yarn.am.memoryOverhead=384
spark\:spark.executor.memory=2688m
spark\:spark.yarn.am.memory=2688m
mapred\:yarn.app.mapreduce.am.command-opts=-Xmx4915m

完全错误：

Py4JJavaError：调用 o4973.fit 时出错。：org.apache.spark.SparkException：作业因阶段失败而中止：阶段 16964.0 中的任务 151 失败 4 次，最近一次失败：阶段 16964.0 中丢失任务 151.3（TID 779444，reco-test-w-0.c.datasetredouteasvendor .internal)：ExecutorLostFailure（其中一个正在运行的任务导致executor 14退出）原因：Executor heartbeat timed out after 175122 ms 驱动程序堆栈跟踪：在 org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1450) 在 org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1438) 在 org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1437) 在 scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59) 在 scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48) 在 org.apache.spark.scheduler.DAGScheduler.abortStage（DAGScheduler.scala:1437）在 org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:811) 在 org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:811) 在 scala.Option.foreach(Option.scala:257) 在 org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:811) 在 org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:1659) 在 org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1618) 在 org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1607) 在 org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:48) 在 org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:632) 在 org.apache.spark.SparkContext.runJob(SparkContext.scala:1871) 在 org.apache.spark.SparkContext.runJob(SparkContext.scala:1884) 在 org.apache.spark.SparkContext.runJob(SparkContext.scala:1897) 在 org.apache.spark.SparkContext.runJob(SparkContext.scala:1911) 在 org.apache.spark.rdd.RDD$$anonfun$collect$1.apply(RDD.scala:893) 在 org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151) 在 org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112) 在 org.apache.spark.rdd.RDD.withScope(RDD.scala:358) 在 org.apache.spark.rdd.RDD.collect(RDD.scala:892) 在 org.apache.spark.rdd.PairRDDFunctions$$anonfun$countByKey$1.apply(PairRDDFunctions.scala:372) 在 org.apache.spark.rdd.PairRDDFunctions$$anonfun$countByKey$1.apply(PairRDDFunctions.scala:372) 在 org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151) 在 org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112) 在 org.apache.spark.rdd.RDD.withScope(RDD.scala:358) 在 org.apache.spark.rdd.PairRDDFunctions.countByKey(PairRDDFunctions.scala:371) 在 org.apache.spark.rdd.RDD$$anonfun$countByValue$1.apply(RDD.scala:1156) 在 org.apache.spark.rdd.RDD$$anonfun$countByValue$1.apply(RDD.scala:1156) 在 org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151) 在 org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112) 在 org.apache.spark.rdd.RDD.withScope(RDD.scala:358) 在 org.apache.spark.rdd.RDD.countByValue(RDD.scala:1155) 在 org.apache.spark.ml.feature.StringIndexer.fit(StringIndexer.scala:91) 在 org.apache.spark.ml.feature.StringIndexer.fit(StringIndexer.scala:66) 在 sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) 在 sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) 在 sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) 在 java.lang.reflect.Method.invoke(Method.java:498) 在 py4j.reflection.MethodInvoker.invoke（MethodInvoker.java:237）在 py4j.reflection.ReflectionEngine.invoke（ReflectionEngine.java:357）在 py4j.Gateway.invoke(Gateway.java:280) 在 py4j.commands.AbstractCommand.invokeMethod(AbstractCommand.java:128) 在 py4j.commands.CallCommand.execute(CallCommand.java:79) 在 py4j.GatewayConnection.run(GatewayConnection.java:211) 在 java.lang.Thread.run(Thread.java:745)

【问题讨论】：

您是否明确设置了spark:spark.executor.cores 和spark:spark.executor.memory 自己？默认情况下，Dataproc 的每个执行程序将拥有多个内核，而且看起来 spark:spark.executor.memory 与 Dataproc 为 n1-highmem-8 计算的默认值不同。
总之，获得更多每个工作单元内存的最简单方法就是调整spark.executor.memory；您甚至可以在提交作业时执行此操作，而无需重建集群；例如，如果从命令行使用 spark-shell 或 pyspark 而不是 Dataproc 作业提交，则运行 pyspark --conf spark.executor.memory=5376m。你应该能够一直提高这个数字，直到你达到大约一台机器的整个尺寸；但是，使用较大的每个执行器内存大小，您将拥有更少的执行器，因此使用较大的内存设置可能会留下一些未使用的核心。
你是对的。 n1-highmem-8 集群的标准是 spark.executor.memory=18619m 和 spark.executor.cores=4。由于我有 8 个 52GB 内存的核心工作人员，我可以设置 spark.executor.memory=50000m 和 spark.executor.cores=8 吗？这么高吗？
一小部分内存是为各种系统守护进程和开销、HDFS、NodeManagers 等保留的，因此您可以查看集群描述 (gcloud dataproc clusters describe <cluster-name>) 中的yarn:yarn.scheduler.maximum-allocation-mb 以找出最大值YARN 分配，并减去spark:spark.yarn.executor.memoryOverhead 以了解您可以在spark.executor.memory 中请求多少。请注意，每个执行程序核心的内存对于 OOM 问题很重要，因此设置 spark.executor.memory=40000m 和 spark.executor.cores=8 将具有与 5000m/cores=1 相同的任务限制
您是否尝试过使用默认设置？默认设置应该比您已经在问题中发布的设置提供更多的内存空间。要深入了解可能的OOM，最简单的方法实际上就是减少spark.executor.cores，同时保持spark.executor.memory不变；默认4核共享18619m；如果您设置为 3 个核心，那么它只会在 3 个核心之间拆分；您可以这样做，直到找到内存和内核的必要平衡，然后根据需要将其调整为单核执行器。

标签： apache-spark apache-spark-ml google-cloud-dataproc

【解决方案1】：

由于这个问题没有答案，总而言之，这个问题似乎与 spark.executor.memory 设置得太低有关，导致执行程序偶尔出现内存不足错误。

建议的解决方法是首先尝试从默认 Dataproc 配置开始，该配置会尝试充分利用实例上的所有可用内核和内存。如果问题仍然存在，则调整spark.executor.memory 和spark.executor.cores 以增加每个任务 的可用内存量（基本上是spark.executor.memory / spark.executor.cores）。

Dennis 还在以下回答中提供了有关 Dataproc 上 Spark 内存配置的更多详细信息：
Google Cloud Dataproc configuration issues

【讨论】：