如何在Spark EMR中通过HeapDumpOnOutOfMemoryError将堆转储至S3

核心结论 #

spark.driver.extraJavaOptions中的-XX:HeapDumpPath不支持直接指定S3路径。原因是JVM的堆转储功能是本地文件系统操作，它无法识别S3的对象存储协议，只能读写本地磁盘路径，因此会抛出"No such file or directory"错误。

EMR环境下直接转储堆文件到S3的可行方案 #

方案1：本地转储+自动同步到S3 #

先将堆转储文件写入EMR节点的本地临时目录，再通过工具自动同步到S3，无需手动登录节点操作。

步骤1：配置堆转储到本地路径

修改Spark配置，指定本地临时目录作为堆转储路径：

// Driver端配置
new SparkConf()
    .set("spark.driver.extraJavaOptions", "-XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/tmp/heap_dumps/driver_heap_dump.hprof")
    // 如果需要监控Executor的堆转储，添加以下配置
    .set("spark.executor.extraJavaOptions", "-XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/tmp/heap_dumps/executor_heap_dump_%p.hprof");

注：%p会替换为进程ID，避免多个Executor的堆转储文件重名。

步骤2：自动同步到S3

有两种自动同步方式：

• 利用EMR日志自动同步：EMR默认会将节点的日志目录（如/var/log/spark/）同步到S3的集群日志路径。可以将堆转储目录软链接到该日志目录，或者修改Spark的spark.log.dir配置指向堆转储目录，让EMR自动同步。
• 添加ShutdownHook自动上传：在Spark作业中添加JVM关闭钩子，当JVM因OOM退出时，自动执行S3上传命令：

// Scala示例，添加到作业初始化代码中
Runtime.getRuntime.addShutdownHook(new Thread(() => {
  val heapDumpPath = "/tmp/heap_dumps/driver_heap_dump.hprof"
  val s3Path = "s3://my-bucket/logs/heapDumps/executor/"
  // 执行aws cli命令上传
  val process = Runtime.getRuntime.exec(s"aws s3 cp $heapDumpPath $s3Path")
  process.waitFor()
}))

方案2：使用Bootstrap Action配置实时监控上传 #

通过EMR的Bootstrap Action在集群启动时部署监控脚本，实时检测本地堆转储目录，一旦生成.hprof文件立即上传到S3。

示例监控脚本（heap_dump_sync.sh）

#!/bin/bash
MONITOR_DIR="/tmp/heap_dumps"
S3_DEST="s3://my-bucket/logs/heapDumps/executor/"

# 创建监控目录
mkdir -p $MONITOR_DIR

# 使用inotifywait实时监控文件创建事件
while true; do
  inotifywait -e create -e moved_to $MONITOR_DIR | while read dir events filename; do
    if [[ $filename == *.hprof ]]; then
      aws s3 cp "$dir$filename" "$S3_DEST"
      # 可选：上传后删除本地文件释放磁盘空间
      rm "$dir$filename"
    fi
  done
done

将该脚本上传到S3，然后在创建EMR集群时添加Bootstrap Action，指定脚本路径即可。

方案3：利用EMR的日志聚合功能 #

如果使用EMR 5.20+版本，可以启用日志聚合功能，它会自动收集所有节点的日志（包括堆转储文件，如果放在指定目录）并上传到S3。只需确保堆转储路径在/var/log/spark/或EMR日志聚合监控的目录下即可。

内容的提问来源于stack exchange，提问作者Danniel_Lee