apache-spark - 如何从 UI 的 Stage DAG 中知道哪行代码导致 OutOfMemory？

Question

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

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

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

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

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

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

Answer 1

由于您不知道哪个转换导致了问题，我建议在转换之间注入(inject)操作，以查看哪个操作会重新创建问题。

这是我用来简化此操作的一个小实用程序功能:
(我称之为 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 中看到确切的行，并且在日志中您将看到在失败之前将发生什么转换。