Spark FileAlreadyExistsException 在写入 JSON 文件时出现阶段失败答案

【问题标题】：Spark FileAlreadyExistsException on stage failure while writing a JSON fileSpark FileAlreadyExistsException 在写入 JSON 文件时出现阶段失败
【发布时间】：2020-07-16 09:52:59
【问题描述】：

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

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

这是我的代码：

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

从火花用户界面，执行器上的错误说

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 次重试都失败，我就会得到驱动程序的真正原因。我已经将模式设置为覆盖，所以这没有帮助。

【问题讨论】：

我也有同样的问题..有时它会起作用，有时它不会..解决这个问题 - 添加代码以在写入之前删除目录..
@Srinivas 在我的情况下不起作用我确保s3_path 在调用.json(s3_path) 之前是唯一的，当写入时任务失败并且路径已经存在时会发生异常。
你在用什么FileOutputCommitter？您可能有兴趣查看hadoop.apache.org/docs/r3.1.1/hadoop-aws/tools/hadoop-aws/…。

标签： apache-spark apache-spark-sql

【解决方案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;

【讨论】：