Spark Structured Streaming读取Delta表遇UnsupportedOperationException问题求助

架构配置 #

Azure Event Hub -> 原始Delta表 -> agg1 Delta表 -> agg2 Delta表

采用Spark Structured Streaming处理数据，通过foreachBatch结合merge操作更新目标Delta表。

报错信息 #

java.lang.UnsupportedOperationException: Detected a data update (for example partKey=ap-2/part-00000-2ddcc5bf-a475-4606-82fc-e37019793b5a.c000.snappy.parquet) in the source table at version 2217. This is currently not supported. If you'd like to ignore updates, set the option 'ignoreChanges' to 'true'. If you would like the data update to be reflected, please restart this query with a fresh checkpoint directory.

核心问题 #

无法以流式方式读取agg1 Delta表，即便将最后一个流的输出从Delta切换为内存，仍会报相同错误，但第一个流（从Event Hub到原始表）运行正常。

备注 #

• agg1 Delta表将日期截断至分钟粒度，agg2 Delta表将日期截断至天粒度
• 关闭所有其他流后，读取agg1的流仍无法正常运行
• agg2 Delta表为全新空表

解决方法 #

1. 启用ignoreChanges选项：若无需感知agg1表的更新操作（仅关注新增数据），在读取agg1的流式查询中添加配置：

spark.readStream
  .format("delta")
  .option("ignoreChanges", "true")
  .load("/path/to/agg1")

该配置会让流式查询忽略源表的更新，仅处理新增批次数据。

2. 重置检查点目录重启查询：若需要agg1表的更新同步到agg2表，必须删除原有检查点目录，使用全新目录重启读取agg1的流式查询。Delta Lake流式读取默认仅跟踪新增数据，源表的更新操作（如merge修改已有数据）无法被旧检查点识别，必须重置检查点从头处理。

3. 调整agg1表写入逻辑：如果agg1表通过merge更新，可考虑改为仅追加模式。比如基于分钟粒度做增量聚合时，避免修改已有分钟粒度的数据，让源表无变更操作，流式读取即可正常运行。

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