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

Question

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

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

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

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

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

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

Answer 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 中看到确切的行，并且在日志中您将看到失败之前将发生什么转换。