如何从 UI 的 Stage DAG 中知道哪行代码导致 OutOfMemory？答案

【问题标题】：How to know what line of code causing OutOfMemory from UI's Stage DAG?如何从 UI 的 Stage DAG 中知道哪行代码导致 OutOfMemory？
【发布时间】：2017-05-07 07:40:03
【问题描述】：

我正在尝试分析我的 spark 作业中的一个问题，该问题因 OutOfMemory 错误而失败。

我正在尝试通过查看 Spark ApplicationMaster 的 UI 来分析问题。 UI 显示作业的阶段和任务细分，我的分析重点是正在经历任务失败的阶段。

Stage UI 有一个 DAG，显示在该阶段执行的实际命令/方法。

问题是我有 150 行代码进行 DataFrame 转换，而最后只有一行代码执行了一项操作 - 保存到镶木地板。这意味着所有“蓝色框”都有最后一行的行号。

是的，他们有一些关于方法的信息（窗口、顺序等），但我在整个代码中都有这些方法，所以我无法知道它指的是哪一行。

我该怎么做才能找到代码的哪一部分有问题？

【问题讨论】：

标签： apache-spark out-of-memory apache-spark-sql

【解决方案1】：

由于您不知道是哪个转换导致了问题，我建议在转换之间注入动作，看看哪个动作会重现问题。

这是我用来简化此操作的一个小实用功能：（我称之为break，因为它破坏了 DAG）

implicit class ExtendedDataFrame(val df : DataFrame) extends AnyVal {   
    def break(name: String = ""):DataFrame = {
      if (name.nonEmpty) {
        println(s"About to break DAG for $name")
      }

      val cached = df.cache()
      cached.count
      cached
    }
}

还有这段代码：

df
    .map(...)
    .map(...)
    .map(...)
    .map(...)
    .action

看起来像这样：

df
    .map(...).break("map 1")
    .map(...).break("map 2")
    .map(...).break("map 3")
    .map(...).break("map 4")
    .action

现在，当您编码失败时，您将在 spark UI 中看到确切的行，并且在日志中您将看到在失败之前将发生什么转换。

【讨论】：

谢谢。我已经有了确切的解决方案，可以在调试时调用show()。我正在寻找不需要中间操作的解决方案。
另外，这涉及到cache()，我担心这会让我的执行者的记忆超负荷。
我同意，但另一方面，不使用 cache() 会强制每个 break 评估之前的转换，这可能需要很长时间
我最终所做的是将每个 DF 存储到 parquet 并立即读取。需要更长的时间，但有助于查明薄弱环节。

【解决方案2】：

我会查看查询的物理计划，也许还会查看 RDD 沿袭。

使用explain 了解物理计划（或df.queryExecution.sparkPlan），这就是将要执行的内容。

可以使用RDD.toDebugString（您可以使用df.rdd 访问）获得RDD 沿袭，它将向您显示阶段和分区（= 任务）。

这样，我会注册一个SparkListener 并开始收听SparkListenerStageSubmitted 和SparkListenerStageCompleted 事件。收听SparkListenerTaskStart 和SparkListenerTaskEnd 也会有所帮助。

所有这些都会让您很好地了解什么以及何时完成或未完成。没有代码审查、更改、重新编译或类似的东西。干净整洁。

可以使用spark.extraListeners Spark 属性注册自定义SparkListener：

应注册的 Spark 侦听器的完全限定类名的逗号分隔列表（当 SparkContext 初始化时）。

见Spark Listeners — Intercepting Events from Spark Scheduler。

【讨论】：