附加模式下的 Spark 2.x Dataframe 写入一致性检查

Question

我正在 for 循环中读取 Spark 中的数据并执行连接并将数据以附加模式写入路径。

for (partition <- partitionlist) {
    var df = spark.read.parquet("path")
    var df2 = df.join(anotherdf, col("col1") === col("col1"))
    df2.write.mode("SaveMode.Append").partitionBy("partitionColumn").format("parquet").save("anotherpath")
}

在上面的示例代码中，我们使用的是 spark 2.X 版本。由于 spark 2 write API 不一致，是否有可能在任何迭代的情况下，如果阶段/任务重试（写入路径）并在几次重试后成功，我们是否有可能看到数据发生重试的那个for循环迭代的写入数据冗余？

编辑：spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version=2 正在使用中。

Answer 1

是的，根据MAPREDUCE-7282MR v2 提交算法应该被弃用，而不是默认算法。因为

v2 MR 提交算法将文件从任务尝试目录移动到 任务提交时的 dest 目录 - 一个接一个

因此它不是原子的

1. 如果任务提交中途失败并且另一个任务尝试 提交 - 除非使用完全相同的文件名，否则输出 第一次尝试可能会包含在最终结果中
2. 如果工人 在任务提交的中途分区，然后在之后继续 已提交另一次尝试，它可能会部分覆盖输出 - 即使文件名相同

有一个对应的 Spark JIRA SPARK-33019，它在 Spark 3 中将提交算法恢复为 v1。