apache-spark - 写入 JSON 文件时在阶段失败时引发 FileAlreadyExistsException

Question

我正在尝试以 JSON 格式将数据帧写入 s3 位置。但是，每当执行程序任务失败并且 Spark 重试该阶段时，它就会抛出 FileAlreadyExistsException。

A similar question之前有人问过，但它使用单独的 spark conf 解决了 ORC 文件，并没有解决我的问题。

这是我的代码:

val result = spark.sql(query_that_OOMs_executor)
result.write.mode(SaveMode.Overwrite).json(s3_path)

在 spark UI 中，执行器上的错误是

ExecutorLostFailure (executor 302 exited caused by one of the running tasks) 
Reason: Container killed by YARN for exceeding memory limits. 4.5 GB of 4.5 GB physical memory used. 
Consider boosting spark.yarn.executor.memoryOverhead or disabling yarn.nodemanager.vmem-check-enabled because of YARN-4714.

但是驱动程序堆栈跟踪显示

Job aborted due to stage failure: Task 1344 in stage 2.0 failed 4 times, most recent failure: Lost task 1344.3 in stage 2.0 (TID 25797, executor.ec2.com, executor 217): org.apache.hadoop.fs.FileAlreadyExistsException: s3://prod-bucket/application_1590774027047/-650323473_1594243391573/part-01344-dc971661-93ef-4abc-8380-c000.json already exists

如何让 spark 尝试覆盖这个 JSON 文件？这样，一旦所有 4 次重试都失败，我就会得到驱动程序的真正原因。我已经将模式设置为覆盖，所以这没有帮助。

Answer 1

出现此问题是因为此处默认使用的 DirectFileOutputCommitter 存在基本问题。

这里有两件事:执行程序 OOM，然后是 FileAlreadyExistsException 重试导致重试(以及 SQL 查询)失败。

原因:DirectFileOutputCommitter 将尝试在单个任务尝试中将输出文件写入最终输出路径。它将通过写入暂存目录然后重命名为最终路径并删除原始路径来实现。这是不好的，容易出现不一致和错误，Spark 也不推荐。

相反，我使用了 the Netflix S3 committer这将以多部分方式执行此操作。它会首先在本地磁盘上写入文件，然后在任务提交期间，每个文件都将分块上传到 S3 但不会立即可见，然后在作业提交期间(只有在所有任务完成时才会发生)成功，这是一个安全的操作)本地磁盘数据将被删除，上传将完成(现在数据将在 S3 上可见)。这可以防止失败的任务直接将内容写入 S3，从而避免 FileAlreadyExistsException 重试。

现在对于执行程序 OOM — 它们仍然在我的查询中发生，但是重试成功了，之前使用 DirectFileOutputCommitter 也失败了。

为了解决这个问题，我基本上是这样做的

set spark.sql.sources.outputCommitterClass=com.netflix.s3.S3DirectoryOutputCommitter;