csv - Spark : avoid task restart when writing

Question

我有一个 Spark 应用程序，可以读取 CSV 并写入 Parquet 文件。

在某些情况下(分配的内存太少、执行器丢失)，Parquet 任务可能会失败并重试；我注意到在这种情况下存在重复的记录，即一些 CSV 在重试时多次写入 Parquet 文件。

避免此类重复的最先进技术是什么？我已经使用 --conf spark.yarn.maxAppAttempts=1 但这仅适用于作业，不适用于任务。如果某一阶段失败，应用程序是否会失败，或者是否有任何方法可以回滚？

Answer 1

Spark 使用 FileOutputCommitter 来管理暂存输出文件和最终输出文件。

FileOutputCommitter 的行为对写入数据的作业的性能有直接影响。它有两个方法:commitTask 和 commitJob。

Apache Spark 2.0 及更高版本使用 Apache Hadoop 2，后者使用 mapreduce.fileoutputcommitter.algorithm.version 的值来控制 commitTask 和 commitJob 的工作方式。

目前 Spark 附带两种默认的 Hadoop 提交算法 - 版本 1 和版本 2。

在版本1中，commitTask 将任务生成的数据从任务临时目录移动到作业临时目录，并且当所有任务完成时，commitJob 将数据从作业临时目录移动到最终目的地。这确保了在作业级别进行事务写入。

在版本2中，commitTask会将任务生成的数据直接移动到最终目的地，而commitJob基本上是一个空操作。这确保了任务级别的事务写入。如果重新提交作业，您可能会看到重复项。

根据您的情况，设置dataframe.write.option("mapreduce.fileoutputcommitter.algorithm.version", "1")以确保在作业级别进行事务写入。

引用:https://databricks.com/blog/2017/05/31/transactional-writes-cloud-storage.html