Spark HistoryServer 介绍与故障排查

HistoryServer 介绍

HistoryServer 是 Spark 内置的一个 HTTP 服务，我们可以通过下面的命令启动它。HistoryServer 启动后，会开启两个线程分别用于解析和清理日志文件。

./sbin/start-history-server.sh

该命令会在 http://<server-url>:8080 上创建一个 Web 界面，用于展示尚未完成以及已经完成的应用与尝试（attempt）。

Spark 作业本身必须配置为记录事件，并将事件日志写入同一个共享的、可写的目录。

例如，如果服务端配置的日志目录为 hdfs://namenode/spark-logs，那么客户端需要设置以下选项：

spark.eventLog.enabled true
spark.eventLog.dir hdfs://namenode/spark-logs

为了更好地介绍 Spark History Server，下面的流程图梳理了它的工作流程。

Spark History Server 依赖于 Spark 的事件（Event）体系。在一个 Spark 任务运行过程中，会产生各种 SparkListenerEvent，例如 ApplicationStart、StageCompleted 等。这些事件会被发送到 ListenerBus，并被已注册的监听器监听。其中，EventLoggingListener 会将这些事件以 JSON 日志文件的形式写入到诸如 HDFS 这样的文件系统中。

History Server 中的 FsHistoryProvider 会周期性地扫描日志存储路径，从日志文件中抽取概要信息（例如 Application_id、user、status、start_time、end_time、event_log_path），并维护一份数据列表。当用户访问 UI 时，系统会在该列表中查找对应的任务，如果存在，就读取相应的事件日志文件以展示详情。

问题

在一些情况下，使用 Spark 提交作业之后，在 Spark HistoryServer WebUI (v3.2.1) 上无法找到对应 appId 的日志信息。

因此，用户无法通过 HistoryServer（HS）来分析作业日志。在分析源码之后，发现问题的根因在于多线程环境下对某个数据结构的错误使用。

数据结构

在解释这个问题之前，需要先了解 HS 中一个关键的数据结构。正是对这个数据结构的误用导致了该问题。HS 使用一个名为 Listing 的数据结构来存储键值对，例如：

<appId, InfoData<appId>>
<logPath, InfoData<logPath>>

日志解析流程

HS 默认每 10 秒运行一次 checkForLogs 方法，以根据日志目录的当前内容来构建应用列表。每次执行时，该方法都会扫描指定的 logDir，并解析日志文件以更新其 KVStore listing。整个流程包含两个主要的方法：

1. 分别更新 / 新增日志文件记录
2. 移除过期的日志文件记录。

这些方法通过 submitLogProcessTask 在不同的线程中执行。

问题场景

当 checkForLogs 首次执行时，用户提交的 ABC 作业仍在运行中。此时日志目录中的文件名为 ABC.inprocess，因此 KVStore Listing 会新增两条记录：

<'ABC', InfoData<'ABC'>>
<'/tmp/ABC.inprocess', InfoData<'/tmp/ABC.inprocess'>>

在下一次运行 checkForLogs 时，ABC 应用已经结束，日志文件 /tmp/ABC.inprocess 已被删除，新的日志文件路径变为 /tmp/ABC，所以在本次 checkForLogs 运行过程中，会启动两个线程并执行如下操作：

线程 1：更新 / 新增日志文件记录，该线程会解析日志路径 /tmp/ABC。首先它会更新 KVStore Listing 中的 <'ABC', InfoData<'ABC'>>，然后会在 KVStore Listing 中新增一条记录 <'/tmp/ABC', InfoData<'/tmp/ABC'>>。

线程 2：移除过期的日志文件记录，该线程会注意到日志路径 /tmp/ABC.inprocess 已经不存在。首先它会将该路径识别为过期数据，然后调用 cleanAppData 方法，在该方法中会执行以下代码：

val app = load(appId)

通常情况下，当该方法加载 appId 时，线程 1 已经完成了对 KVStore Listing 中 <'ABC', InfoData<'ABC'>> 的更新。因此，当下面的代码执行时，isStale 的值会是 false。

val isStale = attempt.headOption.exists { a =>
   if (a.logPath != new Path(logPath).getName()) {
     false
   } else {
     ...
     true
   }
}

然而，在某些情况下，当该方法加载 appId 时，线程 1 还未完成更新。这样一来，当上面的代码执行时，isStale 的值会变为 true。最糟糕的结果是：在线程 1 完成更新之后，该操作又把 KVStore Listing 中的日志文件记录 <'ABC', InfoData<'ABC'>> 删除了。

if (isStale) {
   if (others.nonEmpty) {
     val newAppInfo = new ApplicationInfoWrapper(app.info, others)
     listing.write(newAppInfo)
   } else {
     listing.delete(classOf[ApplicationInfoWrapper], appId)
   }
 }
} catch {
  case _: NoSuchElementException =>
}

因此，日志文件记录 <'ABC', InfoData<'ABC'>> 会被永久删除。当用户在 WebUI 上访问该应用时，即使日志文件本身仍然存在，也只会看到 “Application not found”。

需要注意的是，日志文件记录 <'ABC', InfoData<'ABC'>> 不会再被添加回 KVStore Listing。因为在下一次 checkForLogs 运行时，由于 <'/tmp/ABC', InfoData<'/tmp/ABC'>> 这条记录的存在，表示 ABC 的日志文件无需重新加载，所以该方法会直接跳过这个日志文件。

val info = listing.read(classOf[LogInfo], reader.rootPath.toString())
 
if (shouldReloadLog(info, reader)) {
    ...
    } else {
    false
    }

为了更清晰地展示该 Bug 的触发流程，下面是对应的流程图

解决方案

由于在多线程环境下对 KVStore Listing 的使用存在问题，我们把 cleanAppData 方法变为原子操作，就像 addListing 方法一样，从而让多个线程都可以安全地使用 KVStore Listing，而不会出现数据错误。

例如：

如果线程 1 先于线程 2 获取到 KVStore Listing 的锁，它会更新 <appId, InfoData<appId>> 这条记录，这样线程 2 就可以安全地读取到该记录，进而使 isStale 为 false，cleanAppData 方法也就不会再删除 KVStore Listing 中的记录。

如果线程 2 先于线程 1 获取到 KVStore Listing 的锁，它会使 isStale 为 true，cleanAppData 方法会删除 KVStore Listing 中的记录。但这没关系，因为线程 1 随后会再次向 KVStore Listing 中写入 <appId, InfoData<appId>> 这条记录。

在上述所有情况下，KVStore Listing 中的记录都不会被意外移除。

这就是 Spark v3.2.2 中用来修复该问题的方法（https://github.com/apache/spark/pull/36424/files）。