星火流和增量表: java.lang.UnsupportedOperationException:检测到数据更新

Question

设置：

Azure Event Hub -> raw delta table -> agg1 delta table -> agg2 delta table

数据通过火花结构流进行处理。

对目标增量表的更新是通过foreachBatch使用merge完成的。

结果，我得到了错误：

partKey=ap-2/part-00000-2ddcc5bf-a475-4606-82fc-e37019793b5a.c000.snappy.parquet)：检测到数据更新(例如，源表中的版本2217中的

)。目前不支持这一点。如果您想忽略更新，请将选项“ignoreChanges”设置为“true”。如果希望反映数据更新，请使用新的检查点目录重新启动此查询。

基本上，我无法通过任何类型的流读取agg1增量表。如果我将最后一次流从增量切换到内存，我将得到相同的错误消息。对于第一次流媒体，我没有任何问题。

笔记。

聚合之间的

1. ：agg1增量表(trunc日期到分钟)，agg2增量表(trunc日期到天)。如果关闭所有其他流，最后一个仍然不能工作
2. agg2增量表是新的表，没有数据

。

Answer 1

流是如何在源表上工作的:它读取属于我们源表的文件。它无法处理这些文件中的更改(更新、删除)。如果发生这种情况，您将得到上面的错误。换句话说。DDL操作修改基础文件。唯一的区别是插入。如果配置方式不同，新数据将进入新文件。

要解决这个问题，您需要设置一个选项: ignoreChanges到True。此选项将导致您从修改后的文件中获取所有记录。因此，您将再次获得与以前相同的记录，再加上这个修改后的记录。

问题是:我们有聚合，聚合值存储在检查点中。如果我们再次获得相同的记录(而不是修改的)，我们将识别它为一个更新，我们将增加它的分组键的聚合。

解决方案:我们无法读取agg表来进行其他聚合。我们需要看一下原始的桌子。

参考资料：https://docs.databricks.com/structured-streaming/delta-lake.html#ignore-updates-and-deletes

注意:我正在处理Databricks运行时10.4，所以默认情况下我使用的是新的洗牌合并。