Spark 3.4+DLT流-流左外连接返回空结果问题求助

当modified流为空时，subtracted返回0条记录的本质是流-流左外连接的时间窗口与水位线机制导致的延迟输出：

• Spark的流-流连接要求时间范围条件，目的是限制状态存储的大小，避免无限期保留数据。对于你的左外连接，Spark会将original的记录暂存在状态中，等待是否有符合modTime >= origTime AND modTime <= origTime + 1 hour条件的modified记录出现。
• 只有当origTime的水位线（即origTime + 2 hours）到达后，Spark才会判定不会再有匹配的modified记录，才会将这些未匹配的original记录输出。如果测试时未等待到这个时间点，自然看不到数据。

另外，你设置的1小时时间窗口可能不符合业务实际：modified是original的修改记录，通常modTime会晚于origTime，但1小时的窗口过窄，若修改操作延迟超过1小时，会导致合法的匹配被错误过滤。

方案一：调整流连接参数（适合必须用连接的场景） #

• 缩小水位线等待时间，比如将original的水位线改为10 minutes，测试时等待对应时间后再查看结果：

  original = dlt.readStream("original_table").withWatermark('origTime', '10 minutes')
  modified = dlt.readStream("modified_table").withWatermark('modTime', '15 minutes')
  

• 放宽时间范围条件，比如改为modTime >= origTime，避免因窗口过窄导致的匹配失败：

  subtracted = original.join(modified, expr("""
                   ID = ID_mod AND
                   modTime >= origTime
               """), 'leftOuter') \
               .where(modified['ID_mod'].isNull()).select(original['*'])
  

方案二：使用DLT Merge操作（更贴合业务需求） #

针对"用modified记录替换original对应ID记录"的场景，DLT的Merge（Upsert）操作更直接，无需处理流连接的时间限制：

@dlt.table
def final_records():
    # 读取原始流和修改流
    original_stream = dlt.readStream("original_table")
    modified_stream = dlt.readStream("modified_table")
    
    # 先将原始流的数据作为初始数据插入目标表
    initial_insert = dlt.merge(
        target=dlt.read("final_records"),
        source=original_stream,
        condition="target.ID = source.ID"
    ).whenNotMatchedInsertAll()
    
    # 再用修改流的数据进行更新/插入
    return dlt.merge(
        target=initial_insert,
        source=modified_stream,
        condition="target.ID = source.ID_mod"
    ).whenMatchedUpdateAll()
     .whenNotMatchedInsertAll()

或者更简洁的方式（利用DLT的流处理特性，将原始数据和修改数据合并后做去重）：

@dlt.table
def final_records():
    # 将修改流的ID字段对齐，方便合并
    modified = dlt.readStream("modified_table").withColumnRenamed("ID_mod", "ID")
    # 合并原始流和修改流，修改流的优先级更高
    combined = dlt.readStream("original_table").unionByName(modified, allowMissingColumns=True)
    # 按ID去重，保留最新的记录（假设modTime晚于origTime，取时间最晚的记录）
    return combined.orderBy("modTime", "origTime", ascending=False).dropDuplicates(["ID"])

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