PySpark中Left Outer Join表现为Inner Join问题排查求助

你的左外连接结果看似等同于内连接，核心问题出在流连接未配置水位线以及Append输出模式的特性限制，具体拆解如下：

1. 缺失水位线，Spark无法触发无匹配左表行的输出 #

Spark Streaming左外连接中，对于左表（deals）里没有匹配右表（transactions）的行，Spark需要明确「不会再有后续右表数据能匹配该行」时，才会将其输出。如果不设置水位线（Watermark），Spark会一直等待潜在的匹配数据，永远不会输出这些无匹配的行，最终结果就会和内连接表现一致。

2. 连接条件的时间逻辑合理性不足（非直接报错原因，但影响匹配逻辑） #

你当前的连接条件是用transactions的时间范围匹配deals：

deals.timestamp >= transactions.timestamp AND deals.timestamp <= transactions.timestamp + interval 1 seconds

但左外连接的核心是保留所有deals行，更合理的逻辑应该是以deals的时间为基准，匹配其前后1秒内的transactions，比如：

transactions.timestamp >= deals.timestamp - interval 1 seconds AND transactions.timestamp <= deals.timestamp

3. Append输出模式的固有限制 #

Append模式仅会输出「后续不会被更新」的行。没有水位线的情况下，Spark无法判定左表无匹配行是否还会有后续匹配数据，因此不会将这类行写入结果流。

步骤1：为两个流添加水位线 #

必须基于时间字段为两个流配置水位线，让Spark明确可以清理旧状态的时间点，从而触发无匹配左表行的输出：

# 为deals流设置水位线，延迟时间可根据业务容忍的最大数据延迟调整
deals_with_watermark = deals_parsed.alias("deals").withWatermark("deals.timestamp", "1 minutes")
# 为transactions流设置水位线
transactions_with_watermark = transactions_parsed.alias("transactions").withWatermark("transactions.timestamp", "1 minutes")

步骤2：调整连接条件（优化匹配逻辑） #

以deals时间为基准，匹配时间范围内的transactions：

joined = deals_with_watermark.join(
    transactions_with_watermark,
    expr("""
    deals.symbol = transactions.symbol AND
    transactions.timestamp >= deals.timestamp - interval 1 seconds AND
    transactions.timestamp <= deals.timestamp
    """),
    "leftOuter"
)

步骤3：为窗口操作补充水位线 #

确保窗口聚合能正常清理旧状态，避免内存溢出：

joined_windowed = (
    joined
    .withColumn("window", window(col("deals.timestamp"), "5 minutes"))
    # 窗口水位线需略大于窗口长度，确保所有延迟数据都能被处理
    .withWatermark("window.start", "6 minutes")
)

• 水位线的核心作用是告知Spark：某个时间点之前的所有数据已全部到达，不会再有延迟数据，可以安全清理该时间点前的状态，并输出左外连接中无匹配的行。
• 流处理场景下的左外连接，必须配合水位线才能保证无匹配行的正常输出，否则这类行会被无限期保留在状态中，永远不会出现在结果里。

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