Spark结构化流分组后左外连接报错及二次水印问题咨询

问题背景 #

运行Spark结构化流代码时，对两个分组聚合后的流执行左外连接，触发报错：

Stream-stream LeftOuter join between two streaming DataFrame/Datasets is not supported without a watermark in the join keys, or a watermark on the nullable side and an appropriate range condition

尝试在聚合后的流上添加二次水印解决报错，但添加后流无结果输出。现咨询：

1. 聚合后能否设置二次水印？
2. 如何解决上述连接报错及无输出问题？

原连接代码 #

Dataset<Row> aggStreamA = df
    .withWatermark("dateTime", "2 days")
    .groupBy(
        window(col("dateTime"), windowDuration, slideDuration),
        col(colA)
    )
    .agg(count("*").alias("count_a"));

Dataset<Row> aggStreamB = df
    .withWatermark("dateTime", "2 days")
    .groupBy(
        window(col("dateTime"), windowDuration, slideDuration),
        col(colB)
    )
    .agg(count("*").alias("count_b"));

Dataset<Row> joinedStream = aggStreamA
    .join(
        aggStreamB,
        aggStreamA.col("window.start").equalTo(aggStreamB.col("window.start"))
            .and(aggStreamA.col("window.end").equalTo(aggStreamB.col("window.end")))
            .and(aggStreamA.col(colA).equalTo(aggStreamB.col(colB)))
        ,
        "leftOuter"
    );

StreamingQuery query = joinedStream
    .writeStream()
    .outputMode("append")
    .format("console")
    .start();

query.awaitTermination();

报错信息 #

Stream-stream LeftOuter join between two streaming DataFrame/Datasets is not supported without a watermark in the join keys, or a watermark on the nullable side and an appropriate range condition

尝试的修改代码 #

Dataset<Row> aggStreamB = df
    .withWatermark("dateTime", "2 days")
    .groupBy(
        window(col("dateTime"), windowDuration, slideDuration),
        col(colB)
    )
    .agg(count("*").alias("count_b")).withColumn("start", col("window.start"))
    .withColumn("start", col("window.start"))
    .withWatermark("start", "0 second");

Dataset<Row> joinedStream = aggStreamA.as("a")
    .join(
        aggStreamB.as("b"),
        aggStreamA.col("window.start").equalTo(aggStreamB.col("window.start"))
            .and(aggStreamA.col("window.end").equalTo(aggStreamB.col("window.end")))
            .and(aggStreamA.col(colA).equalTo(aggStreamB.col(colB)))
            .and(expr("a.start >= b.start and a.start <= b.start + interval 2 days"))
        ,
        "leftOuter"
    );

StreamingQuery query = joinedStream
    .writeStream()
    .outputMode("append")
    .format("console")
    .start();

query.awaitTermination();

二次水印无输出的对比代码 #

有输出的代码 #

// 有输出的代码
Dataset<Row> aggStreamA = df
    .withWatermark("dateTime", "2 days")
    .groupBy(
        window(col("dateTime"), windowDuration, slideDuration),
        col(colA)
    )
    .agg(count("*").alias("count_a"))
    .withColumn("start", col("window.start"));

StreamingQuery query = aggStreamA
    .writeStream()
    .outputMode("append")
    .format("console")
    .start();

query.awaitTermination();

无输出的代码 #

// 无输出的代码
Dataset<Row> aggStreamA = df
    .withWatermark("dateTime", "2 days")
    .groupBy(
        window(col("dateTime"), windowDuration, slideDuration),
        col(colA)
    )
    .agg(count("*").alias("count_a"))
    .withColumn("start", col("window.start"))
    .withWatermark("start", "0 second");

StreamingQuery query = aggStreamA
    .writeStream()
    .outputMode("append")
    .format("console")
    .start();

query.awaitTermination();

问题解答 #

1. 聚合后能否设置二次水印？ #

可以，但要求水印列具备明确的事件时间语义（需从原始流的事件时间衍生，符合时间推进逻辑）。你遇到的无输出问题核心原因：

• 聚合后的append模式流，仅当窗口超出原始水印阈值（2天）时才会输出结果；
• 添加的withWatermark("start", "0 second")中，start是固定的窗口起始时间，0秒水印会让Spark立即清理所有“旧数据”，而聚合后的窗口数据还未到输出时机就被过滤，导致无输出。

2. 解决连接报错及无输出问题 #

Spark流-流左外连接要求满足以下任一条件：

• 连接键上配置水印，且两端流均设置水印；
• nullable侧（左外连接的右表）配置水印，同时添加事件时间范围条件。

结合窗口聚合场景，正确方案如下：

修正代码示例

// 处理流A：提取窗口起始时间作为事件时间，设置与原始水印一致的二次水印
Dataset<Row> aggStreamA = df
    .withWatermark("dateTime", "2 days")
    .groupBy(
        window(col("dateTime"), windowDuration, slideDuration),
        col(colA)
    )
    .agg(count("*").alias("count_a"))
    .withColumn("event_time", col("window.start"))
    .withWatermark("event_time", "2 days");

// 处理流B：同样配置二次水印
Dataset<Row> aggStreamB = df
    .withWatermark("dateTime", "2 days")
    .groupBy(
        window(col("dateTime"), windowDuration, slideDuration),
        col(colB)
    )
    .agg(count("*").alias("count_b"))
    .withColumn("event_time", col("window.start"))
    .withWatermark("event_time", "2 days");

// 左外连接：窗口匹配+键匹配+事件时间范围条件
Dataset<Row> joinedStream = aggStreamA.as("a")
    .join(
        aggStreamB.as("b"),
        expr("a.window.start = b.window.start")
            .and(expr("a.window.end = b.window.end"))
            .and(expr("a." + colA + " = b." + colB))
            .and(expr("b.event_time between a.event_time - interval 2 days and a.event_time + interval 2 days")),
        "leftOuter"
    );

// append模式下，需等待窗口超过水印阈值才会输出，测试时可缩小水印时长快速验证
StreamingQuery query = joinedStream
    .writeStream()
    .outputMode("append")
    .format("console")
    .start();

query.awaitTermination();

关键说明

• 二次水印配置：用窗口起始时间作为事件时间列，水印时长与原始流保持一致，既满足Spark连接要求，又不会提前过滤未输出的窗口数据；
• 连接条件：必须添加事件时间范围条件，Spark依赖该条件确定右表数据的匹配窗口，避免无限等待；
• 输出时机：append模式下，聚合窗口需超过水印阈值才会输出，测试时可将水印改为1分钟，快速验证输出效果。

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