Databricks中Python结合Autoloader实现CDC无法捕获新增行报错问题

问题表现 #

• 基于Python调用Autoloader实现CDC逻辑时，无法捕获DataFrame新增行，公开资料仅能找到SQL实现方案，无Python实现参考
• 作业运行抛出java.lang.UnsupportedOperationException异常，提示：检测到源表版本4存在数据更新，当前不支持该场景，可设置'ignoreChanges'为true忽略更新，或使用全新检查点目录重启查询以同步更新
• 尝试修改相关存储路径后报错仍存在，无法正常捕获新增数据

触发原因 #

该异常是Autoloader默认文件监听逻辑的原生限制：默认配置下不支持已处理过的源文件发生内容变更。仅修改存储路径后报错不消失，基本是两类问题：

1. 没有在Python读流配置中正确传入ignoreChanges参数
2. 更换路径时没有同步更换全新的空检查点目录，Autoloader仍读取旧检查点中存储的源版本记录，持续触发异常

正确Python配置方式 #

所有SQL实现支持的参数，在Python API中都可以通过.option()方法直接传入键值对，不需要特殊转换，按以下步骤配置即可解决：

1. 读流配置中显式开启ignoreChanges，同时指定全新的schema存储路径，不要复用旧路径

# 云文件源Autoloader配置示例
source_df = spark.readStream.format("cloudFiles") \
  .option("cloudFiles.format", "parquet") \
  .option("cloudFiles.schemaLocation", "abfss://<容器>@<存储账号>.dfs.core.windows.net/autoloader/schema/new_v1") \
  .option("ignoreChanges", "true") \
  .load("abfss://<容器>@<存储账号>.dfs.core.windows.net/source/path")

注：cloudFiles.format参数需替换为实际源数据格式，支持csv/avro/json/delta等

2. 写流时必须指定完全未使用过的空检查点目录，禁止复用历史报错作业的检查点路径

stream_query = source_df.writeStream \
  .format("delta") \
  .outputMode("append") \
  .option("checkpointLocation", "abfss://<容器>@<存储账号>.dfs.core.windows.net/autoloader/checkpoint/new_v1") \
  .start("abfss://<容器>@<存储账号>.dfs.core.windows.net/target/path")

3. 如果源是开启了变更数据馈送的Delta表，直接使用Delta原生读流能力做CDC即可，不需要走cloudFiles文件监听逻辑，不会触发该类异常

# Delta表CDC读取示例
delta_cdc_df = spark.readStream.format("delta") \
  .option("readChangeFeed", "true") \
  .option("startingVersion", 4) \
  .table("<源Delta表全限定名>")

注：startingVersion参数需替换为实际需要起始消费的版本号

注意事项 #

• ignoreChanges开启后，Autoloader会重新处理发生变更的源文件，需要配合源数据中的主键、变更时间戳字段做去重逻辑，避免数据重复写入
• 检查点目录存储了Autoloader所有消费进度、schema变更、源版本记录，只要目录中存在历史残留数据，新配置就可能不生效，首次重启修复前建议确认对应路径为空
• 不要手动修改检查点目录下的文件内容，会直接导致流作业状态损坏

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