Spark写入HDFS时如何实现幂等性保障？

嘿，这个问题问到点子上了——这可是Spark生产环境里绕不开的稳定性问题，我结合实际经验给你拆解清楚：

首先回应你的初始猜测：默认情况下，Spark故障重试确实有概率导致重复写入，但不是所有场景都会发生，得结合执行机制具体分析。

场景1：写入单个数据项时发生故障 #

你的担心其实有点多余，因为Spark写Parquet到HDFS的流程是做了原子性保障的：

• 单个任务写Parquet时，会先把数据写到临时文件（比如命名格式是part-r-xxxx-xxxx-xxxx-xxxx.inprogress这类）；
• 只有当任务完全执行成功，才会把临时文件原子性重命名为正式的Parquet文件（比如part-r-xxxx-xxxx-xxxx-xxxx.parquet）；
• 如果任务中途故障，临时文件会被Spark的清理机制（或HDFS的过期策略）自动删除，重试时会重新生成临时文件再尝试写入。

HDFS的重命名操作是原子性的——要么完全成功，要么完全失败，不存在“半成功”的状态。所以这种场景下，重复写入的概率几乎为0，不用过度担心。

场景2：DAG重启导致部分任务已完成 #

这里要分两种情况讨论：

情况A：单个任务失败的局部重试 #

Spark会跟踪每个任务的执行状态，当某个Stage里的部分任务失败时，Driver只会重试失败的任务，已经成功完成的任务（包括它们的输出文件）会被标记为“已完成”，不会重新执行。这是因为Spark依赖RDD的Lineage（血统）机制，只会重新计算失败的部分，不会从头跑整个DAG。这种情况下，不会出现重复写入。

情况B：整个Application重启（Driver故障） #

如果是Driver节点挂了，集群管理器（比如YARN、K8s）重启整个Application，这时候Driver之前记录的任务状态会丢失，Spark会重新执行整个DAG。这时候如果之前已经有部分写入任务完成，就会导致重复写入——这才是需要重点防范的场景。

如何实现HDFS输出的幂等性？ #

针对上面的风险，这里有几种生产环境常用的方案：

1. 利用Spark内置的输出模式 #

• overwrite模式：最直接的方案，写入前会清空目标目录（或指定分区），不管之前有没有数据。如果是分区表，Spark 2.3+支持配置spark.sql.sources.partitionOverwriteMode = dynamic，开启后只会覆盖你写入的特定分区，不会影响其他分区的数据。

  // 全局覆盖整个目录
  df.write.mode("overwrite").parquet("/target/path")
  
  // 动态覆盖指定分区
  spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")
  df.write.mode("overwrite").partitionBy("dt").parquet("/target/partitioned-path")
  

• ignore模式：如果目标目录已存在，直接跳过写入。适合“只写一次”的场景，重试时不会重复写入，但无法更新已有数据。

2. 临时目录+原子移动（手动实现强幂等） #

核心思路是：先把数据写到一个唯一的临时目录，确认所有写入任务完全成功后，再原子性地把临时目录移动到目标路径。中途故障的话，临时目录可以直接清理，重试时重新生成即可。

import org.apache.hadoop.fs.{FileSystem, Path}

// 生成唯一临时目录（用UUID避免冲突）
val tempDir = s"/tmp/spark-output-${java.util.UUID.randomUUID()}"
// 写入临时目录
df.write.parquet(tempDir)

// 确认写入完成后，执行原子移动
val hadoopConf = spark.sparkContext.hadoopConfiguration
val fs = FileSystem.get(hadoopConf)
val targetPath = new Path("/target/output")

// 如果目标路径已存在，先删除（可选，根据业务需求）
if (fs.exists(targetPath)) fs.delete(targetPath, true)
// HDFS的rename是原子操作
fs.rename(new Path(tempDir), targetPath)

3. 开启Hive ACID事务（适合Parquet表） #

如果是用Spark SQL写入Hive管理的Parquet表，可以开启Hive的ACID事务支持，让写入操作具备原子性、一致性、隔离性——失败自动回滚，成功才会提交，完全避免重复或部分写入。
需要先配置Spark参数：

spark.conf.set("spark.sql.hive.convertMetastoreParquet", "false")
spark.conf.set("spark.sql.hive.metastorePartitionPruning", "true")

然后创建表时指定事务属性：

CREATE TABLE my_table (id INT, name STRING)
PARTITIONED BY (dt STRING)
STORED AS PARQUET
TBLPROPERTIES ('transactional'='true');

之后写入数据时，Spark会自动以事务方式提交，重试不会导致重复。

4. 基于时间/业务键的分区覆盖 #

如果你的数据是按时间（比如dt、hour）或唯一业务键分区，每次写入只针对特定分区，用overwrite模式覆盖该分区即可。比如每天写入当天的dt=20240520分区，就算重试多次，最终也只会保留一份完整的分区数据，不会产生重复。

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