如何实现Spark批处理作业在ADLS Gen2上的全局提交？

以下是几个可直接使用的内置/生态级解决方案，解决部分写入和数据重复问题：

1. Spark原生原子写入配置（无额外依赖） #

Spark 3.0+ 原生支持云存储的原子写入，配合Hadoop 3.2+版本（ADLS Gen2的完整原子操作支持从Hadoop 3.2开始），只需确保以下配置：

• 显式开启原子写入：spark.sql.files.atomicWrite=true（默认已开启，可在作业提交时显式指定）
• 使用Hadoop 3.2及以上版本的ADLS Gen2客户端

这个配置会让Spark将所有任务的输出先写入作业级临时目录，只有当整个批处理作业的所有任务都成功完成后，才会通过ADLS Gen2的原子重命名操作，将临时目录下的所有文件一次性移动到目标路径。只要有任一任务失败，临时目录的内容不会被提交，彻底避免部分数据残留。

2. Delta Lake（Spark生态内置ACID解决方案） #

Delta Lake是Spark官方生态的扩展，提供开箱即用的ACID事务支持，完美适配ADLS Gen2：

• 写入时自动保证原子性：只有整个作业成功完成，数据才会被提交到Delta表；作业失败时，不会留下任何部分数据
• 自动处理重复提交：Delta Lake会跟踪事务ID，重试作业时不会产生重复数据
• 支持版本回溯、增量查询等高级特性，适合大规模数据集的长期存储

使用方式只需将DataFrame写入Delta格式：

df.write
  .format("delta")
  .mode("append") // 或"overwrite"，根据业务需求选择
  .save("abfss://container@storageaccount.dfs.core.windows.net/target-path")

3. 基于ADLS Gen2原子重命名的手动可靠方案 #

如果不想引入额外依赖，可利用ADLS Gen2的目录原子重命名特性（该操作是原子性的，要么全成功要么全失败），结合Spark的临时写入策略：

1. 将数据写入一个唯一的临时目录（比如包含作业ID的路径：abfss://container@storageaccount.dfs.core.windows.net/temp/job-xxxx）
2. 作业成功完成后，通过Spark内置的Hadoop FileSystem API执行原子重命名，将临时目录移动到目标路径
3. 作业失败时直接丢弃临时目录（无需回滚目标路径）

示例代码（Scala）：

import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.sql.SparkSession

val spark = SparkSession.active
val fs = FileSystem.get(spark.sparkContext.hadoopConfiguration)

// 写入临时目录
val tempPath = new Path("abfss://container@storageaccount.dfs.core.windows.net/temp/job-1234")
df.write.parquet(tempPath.toString)

// 原子重命名到目标路径
val targetPath = new Path("abfss://container@storageaccount.dfs.core.windows.net/target")
if (fs.exists(targetPath)) {
  // 可选：如果需要覆盖，先删除目标（但原子重命名本身可覆盖，取决于ADLS配置）
  fs.delete(targetPath, true)
}
fs.rename(tempPath, targetPath)

方案选择建议 #

• 优先选Spark原生原子写入：无额外依赖，配置简单，适合已有Spark作业的快速改造
• 需要事务、版本控制选Delta Lake：适合长期维护的大数据湖场景，后续查询和数据管理更高效
• 轻量需求选ADLS原子重命名：手动但可靠，适合不想修改太多作业逻辑的场景

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