收到 TimeoutException 的可能原因是啥：使用 Spark 时，期货在 [n 秒] 后超时 [重复]

Posted 2023-04-15

技术标签:

【中文标题】收到 TimeoutException 的可能原因是啥：使用 Spark 时，期货在 [n 秒] 后超时 [重复]【英文标题】：What are possible reasons for receiving TimeoutException: Futures timed out after [n seconds] when working with Spark [duplicate]收到 TimeoutException 的可能原因是什么：使用 Spark 时，期货在 [n 秒] 后超时 [重复] 【发布时间】：2016-11-07 20:31:50 【问题描述】：

我正在开发 Spark SQL 程序，但收到以下异常：

16/11/07 15:58:25 ERROR yarn.ApplicationMaster: User class threw exception: java.util.concurrent.TimeoutException: Futures timed out after [3000 seconds]
java.util.concurrent.TimeoutException: Futures timed out after [3000 seconds]
    at scala.concurrent.impl.Promise$DefaultPromise.ready(Promise.scala:219)
    at scala.concurrent.impl.Promise$DefaultPromise.result(Promise.scala:223)
    at scala.concurrent.Await$$anonfun$result$1.apply(package.scala:190)
    at scala.concurrent.BlockContext$DefaultBlockContext$.blockOn(BlockContext.scala:53)
    at scala.concurrent.Await$.result(package.scala:190)
    at org.apache.spark.sql.execution.joins.BroadcastHashJoin.doExecute(BroadcastHashJoin.scala:107)
    at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$5.apply(SparkPlan.scala:132)
    at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$5.apply(SparkPlan.scala:130)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:150)
    at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:130)
    at org.apache.spark.sql.execution.Project.doExecute(basicOperators.scala:46)
    at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$5.apply(SparkPlan.scala:132)
    at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$5.apply(SparkPlan.scala:130)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:150)
    at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:130)
    at org.apache.spark.sql.execution.Union$$anonfun$doExecute$1.apply(basicOperators.scala:144)
    at org.apache.spark.sql.execution.Union$$anonfun$doExecute$1.apply(basicOperators.scala:144)
    at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:245)
    at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:245)
    at scala.collection.immutable.List.foreach(List.scala:381)
    at scala.collection.TraversableLike$class.map(TraversableLike.scala:245)
    at scala.collection.immutable.List.map(List.scala:285)
    at org.apache.spark.sql.execution.Union.doExecute(basicOperators.scala:144)
    at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$5.apply(SparkPlan.scala:132)
    at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$5.apply(SparkPlan.scala:130)
    at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:150)
    at org.apache.spark.sql.execution.SparkPlan.execute(SparkPlan.scala:130)
    at org.apache.spark.sql.execution.columnar.InMemoryRelation.buildBuffers(InMemoryColumnarTableScan.scala:129)
    at org.apache.spark.sql.execution.columnar.InMemoryRelation.<init>(InMemoryColumnarTableScan.scala:118)
    at org.apache.spark.sql.execution.columnar.InMemoryRelation$.apply(InMemoryColumnarTableScan.scala:41)
    at org.apache.spark.sql.execution.CacheManager$$anonfun$cacheQuery$1.apply(CacheManager.scala:93)
    at org.apache.spark.sql.execution.CacheManager.writeLock(CacheManager.scala:60)
    at org.apache.spark.sql.execution.CacheManager.cacheQuery(CacheManager.scala:84)
    at org.apache.spark.sql.DataFrame.persist(DataFrame.scala:1581)
    at org.apache.spark.sql.DataFrame.cache(DataFrame.scala:1590)
    at com.somecompany.ml.modeling.NewModel.getTrainingSet(FlowForNewModel.scala:56)
    at com.somecompany.ml.modeling.NewModel.generateArtifacts(FlowForNewModel.scala:32)
    at com.somecompany.ml.modeling.Flow$class.run(Flow.scala:52)
    at com.somecompany.ml.modeling.lowForNewModel.run(FlowForNewModel.scala:15)
    at com.somecompany.ml.Main$$anonfun$2.apply(Main.scala:54)
    at com.somecompany.ml.Main$$anonfun$2.apply(Main.scala:54)
    at scala.Option.getOrElse(Option.scala:121)
    at com.somecompany.ml.Main$.main(Main.scala:46)
    at com.somecompany.ml.Main.main(Main.scala)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
    at org.apache.spark.deploy.yarn.ApplicationMaster$$anon$2.run(ApplicationMaster.scala:542)
16/11/07 15:58:25 INFO yarn.ApplicationMaster: Final app status: FAILED, exitCode: 15, (reason: User class threw exception: java.util.concurrent.TimeoutException: Futures timed out after [3000 seconds])

我从堆栈跟踪中识别出的代码的最后一部分是com.somecompany.ml.modeling.NewModel.getTrainingSet(FlowForNewModel.scala:56)，这让我进入了这一行：profilesDF.cache() 在缓存之前，我在 2 个数据帧之间执行联合。我已经看到了一个关于在加入之前保留两个数据帧的答案here 我仍然需要缓存联合数据帧，因为我在我的几个转换中使用它

我想知道什么可能导致抛出这个异常？搜索它让我找到了处理 rpc 超时异常或一些安全问题的链接，这不是我的问题如果您对如何解决它也有任何想法，我显然会很感激，但即使只是理解问题也会帮助我解决它

提前致谢

【问题讨论】：

【参考方案1】：

问题：我想知道什么可能导致抛出此异常？

答案：

spark.sql.broadcastTimeout 300 Timeout in seconds for the broadcast wait time in broadcast joins

spark.network.timeout 120s 所有网络交互的默认超时时间.. spark.network.timeout (spark.rpc.askTimeout), spark.sql.broadcastTimeout, spark.kryoserializer.buffer.max(如果你使用的是 kryo 序列化）等使用大于默认值的值进行调整为了处理复杂的查询。您可以从这些值开始根据您的 SQL 工作负载进行相应调整。

注意：Doc says that

以下选项（参见 spark.sql. 属性）也可用于调整查询执行的性能。这些选项可能会在未来的版本中被弃用，因为更多的优化会自动执行。*

此外，为了更好地理解，您可以查看BroadCastHashJoin，其中执行方法是上述堆栈跟踪的触发点。

protected override def doExecute(): RDD[Row] = 
    val broadcastRelation = Await.result(broadcastFuture, timeout)

    streamedPlan.execute().mapPartitions  streamedIter =>
      hashJoin(streamedIter, broadcastRelation.value)

【讨论】：

【参考方案2】：

很高兴知道 Ram 的建议在某些情况下有效。我想提一下，我偶然发现了这个异常几次（包括描述的here）。

大部分时间，这是由于某些执行程序上几乎无声的 OOM。检查 SparkUI 是否有失败的任务，该表的最后一列：您可能会注意到 OOM 消息。

如果对 spark 内部结构了解得很好，广播的数据会通过驱动程序。所以驱动程序有一些线程机制来收集来自执行程序的数据，并将其发送回所有人。如果在某些时候执行程序失败，您可能会遇到这些超时。

【讨论】：

【参考方案3】：

我在向Yarn-cluster提交作业时设置了master as local[n]。

在集群上运行时不要在代码中设置master，而是使用--master。

【讨论】：

你能解释一下为什么会出现这个问题吗？【参考方案4】：

如果您启用了 dynamicAllocation，请尝试禁用此配置 (spark.dynamicAllocation.enabled=false)。您可以在 conf/spark-defaults.conf 下设置此 spark 配置，如 --conf 或在代码中。