本地Spark无Databricks创建ADLS Gen2 Delta表遇版本不连续错误

这个错误本质是Delta日志的版本记录出现重复或损坏，结合你的配置和场景，核心问题集中在以下几点：

1. 日志存储类配置错误 #

你使用了S3SingleDriverLogStore，这是专为AWS S3设计的日志存储实现，在Azure ADLS Gen2环境下必须替换为AzureLogStore，否则会导致日志写入逻辑混乱，生成重复的版本号。

2. 重复的spark.jars配置导致依赖异常 #

多次调用.config("spark.jars", ...)会覆盖之前的配置，导致部分依赖jar未被加载或版本冲突。应将所有需要的jar用逗号分隔合并到同一个spark.jars配置中，更推荐用spark.jars.packages自动管理依赖，避免手动下载带来的版本不兼容问题。

3. 认证配置冲突 #

你同时配置了针对Blob存储的SharedKey和针对ADLS Gen2的OAuth两种认证方式，这会导致文件系统客户端逻辑混乱，需清理SharedKey相关配置，仅保留abfss协议所需的OAuth认证。

4. 目标路径存在损坏的Delta日志 #

若之前写入失败过，ADLS Gen2目标路径下可能残留了损坏的_delta_log目录，其中存在重复的版本0文件，需先手动删除该目录。

修正后的完整代码示例 #

from pyspark.sql import SparkSession
from delta.tables import *

# 替换为你的实际参数
storage_account_name = "your-storage-account"
container_name = "your-container"
client_id = "your-client-id"
client_secret = "your-client-secret"
directory_id = "your-tenant-id"

spark = SparkSession.builder.master("local[*]")\
    .config("spark.jars.packages", "io.delta:delta-core_2.12:2.3.0,org.apache.hadoop:hadoop-azure:3.3.1,com.microsoft.azure:azure-storage:8.6.6")\
    .config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension")\
    .config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog")\
    # 替换为ADLS Gen2专属日志存储类
    .config("spark.delta.logStore.class", "org.apache.spark.sql.delta.storage.AzureLogStore")\
    .config("spark.driver.memory", "4g")\
    .config("spark.executor.memory", "4g")\
    # 合并所有需要的jar配置，用逗号分隔
    .config("spark.jars", r"C:\Users\Proem Sports\Documents\Jupyter notebooks\jars\mysql-connector-j-8.0.31.jar")\
    .config("spark.sql.legacy.parquet.int96RebaseModeInRead", "CORRECTED")\
    .config("spark.sql.legacy.parquet.int96RebaseModeInWrite", "CORRECTED")\
    .config("spark.sql.legacy.parquet.datetimeRebaseModeInRead", "CORRECTED")\
    .config("spark.sql.legacy.parquet.datetimeRebaseModeInWrite", "CORRECTED")\
    .config("spark.sql.execution.arrow.pyspark.enabled", "true")\
    .config("spark.sql.legacy.timeParserPolicy", "CORRECTED")\
    .config("spark.sql.warehouse.dir", r"C:\Users\Proem Sports\Documents\Jupyter notebooks\Dev_scripts\metastore_db")\
    .config("spark.delta.commitInfo.merge.enabled", "true")\
    # 移除冲突的SharedKey认证配置
    .enableHiveSupport()\
    .getOrCreate()

# 配置ADLS Gen2的OAuth认证
spark.conf.set(f"fs.azure.account.auth.type.{storage_account_name}.dfs.core.windows.net", "OAuth")
spark.conf.set(f"fs.azure.account.oauth.provider.type.{storage_account_name}.dfs.core.windows.net", "org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider")
spark.conf.set(f"fs.azure.account.oauth2.client.id.{storage_account_name}.dfs.core.windows.net", client_id)
spark.conf.set(f"fs.azure.account.oauth2.client.secret.{storage_account_name}.dfs.core.windows.net", client_secret)
spark.conf.set(f"fs.azure.account.oauth2.client.endpoint.{storage_account_name}.dfs.core.windows.net", f"https://login.microsoftonline.com/{directory_id}/oauth2/token")

# 读取CSV文件
df = spark.read.format("csv")\
    .load(f"abfss://{container_name}@{storage_account_name}.dfs.core.windows.net/RAW_DATA/MERHCHANDISE/MERCH_20230424_.csv", 
          header=True, inferSchema=True)

# 写入Delta表（需覆盖已有数据可添加.mode("overwrite")）
df.write.format("delta")\
    .option("overwriteSchema", "true")\
    .option('delta.columnMapping.mode', 'name')\
    .save(f"abfss://{container_name}@{storage_account_name}.dfs.core.windows.net/tables/delta_table1")

额外注意事项 #

• 先确认ADLS Gen2目标路径tables/delta_table1下无残留的_delta_log目录，若之前写入失败过，需在Azure门户或Azure CLI中手动删除。
• Spark 3.3.0搭配Delta 2.3.0是官方推荐的兼容组合，无需更换版本。
• 优先使用spark.jars.packages自动管理依赖，减少手动下载jar带来的版本冲突风险。

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