流-流左外连接已加水印仍报错，是否需配置apply_changes？

报错原因 #

流-流左外连接的核心要求是参与连接的两个数据流都必须配置水印，你当前仅为右侧流（DAC）添加了水印，左侧流（ACT）未设置任何水印，这违反了Spark流处理的状态管理规则，因此触发报错。

Spark需要通过水印来确定何时可以清理过期的连接状态数据，若只有一侧有水印，系统无法判断左侧流的旧数据是否还需要保留以匹配未来的右侧流数据，也就无法安全执行左外连接。

修复步骤 #

1. 为左侧流添加水印
   找到左侧流（activity_silver）中的事件时间字段（比如和右侧流一致的_fivetran_synced），添加withWatermark配置，水印延迟时间可根据业务场景调整。

2. （推荐）补充时间范围连接条件
   在连接条件中加入基于事件时间的范围约束，帮助Spark缩小需要维护的状态窗口，提升性能，同时符合多数流场景下的业务逻辑（仅关联相近时间内的事件）。

修改后的代码示例 #

@dlt.view
def vw_ix_f_activity_gold():
    return (
        spark.readStream
        .option("readChangeFeed", "true")
        .table("lakehouse_poc.poc_streaming.activity_silver")
        # 为左侧流添加水印
        .withWatermark("_fivetran_synced", "5 seconds")
        .alias("ACT")
        
        # Join with Oracle Activity Data
        .join(
            spark.readStream.table("lakehouse_poc.poc_streaming.ix_d_activity_gold")
            .withWatermark("_fivetran_synced", "5 seconds")
            .alias("DAC"),
            # 组合关联键与时间范围约束
            [
                F.col("ACT.activity_seq") == F.col("DAC.BK_activity_seq"),
                # 限定时间窗口，可根据业务调整区间
                F.col("ACT._fivetran_synced") >= F.col("DAC._fivetran_synced") - F.expr("interval 5 seconds"),
                F.col("ACT._fivetran_synced") <= F.col("DAC._fivetran_synced") + F.expr("interval 5 seconds")
            ],
            "left"
        )
        # Select and rename columns
        .select(
            ...
    )

dlt.create_streaming_table(
    name = "ix_f_activity_gold",
)

dlt.apply_changes(
    target = "ix_f_activity_gold",
    source = "vw_ix_f_activity_gold",
    keys = ["BK_activity_seq"],
    sequence_by = "_fivetran_synced",
    stored_as_scd_type = 1
)

额外说明 #

• apply_changes部分无需额外配置，问题根源完全在于流连接阶段的水印缺失，和SCD类型1的应用逻辑无关。
• 水印延迟时间需根据业务数据的最大延迟情况设置，确保不会丢失需要关联的事件。

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