Databricks中Watermark与DataFrame.groupBy配合失效问题排查

处理一对多关联场景时，尝试使用DataFrame.groupBy()进行流聚合，代码如下：

@dlt.table(name = silver_table_name)
def silver():
    appointmentsDf = spark.readStream.table(f"{bronze_catalog_name}.{bronze_schema_name}.{bronze_appointments_table_name}")
    # TODO doesn't work
    # appointmentsDf = appointmentsDf.withWatermark("inserted_datetime", "1 hour")

    answersDf = spark.readStream.table(f"{bronze_catalog_name}.{bronze_schema_name}.{bronze_answers_table_name}")
    # TODO - doesn't work
    # answersDf = answersDf.withWatermark("answer_inserted_datetime", "1 hour")

    df = appointmentsDf.join(answersDf, appointmentsDf["id"] == answersDf["appointmentId"], "inner")
    # TODO - doesn't work
    # df = df.withWatermark("inserted_datetime", "1 hour")

    df = df.groupBy('appointmentId') \
        .agg(max(when(col("prompt") == lit(QUESTION_BALL_FITTING), 'answer').otherwise(None)).alias('answer'))

运行后出现错误：

com.databricks.pipelines.common.errors.DLTAnalysisException: Failed to start stream inno_gametime_data_204615620297433 in either append mode or complete mode.
Append mode error: Append output mode not supported when there are streaming aggregations on streaming DataFrames/DataSets without watermark;

即使取消注释任意或全部.withWatermark()语句，仍会出现完全相同的错误。

核心错误原因 #

流聚合使用Append模式时，Spark需要明确知道何时可以确定某条聚合键的结果不会再被后续数据更新，仅单独添加watermark无法满足要求，还需要配合以下条件：

1. 两个流表关联时，必须配置基于watermark的延迟容忍策略，让Spark能够清理过期的关联状态
2. 聚合操作的watermark需要和聚合逻辑配合，确保Spark可以判断聚合结果已“最终确定”

修正步骤 #

1. 给两个流表分别添加watermark，并在join时指定时间范围关联条件（可根据业务调整时间范围），让Spark能确定状态清理时机
2. 关联后生成统一事件时间字段，并基于该字段添加watermark
3. 调整聚合逻辑，让Spark可以通过watermark判断聚合结果是否已最终定型

修正后的代码示例 #

from pyspark.sql import functions as F

@dlt.table(name = silver_table_name)
def silver():
    # 给预约流添加watermark
    appointmentsDf = spark.readStream.table(f"{bronze_catalog_name}.{bronze_schema_name}.{bronze_appointments_table_name}") \
        .withWatermark("inserted_datetime", "1 hour")

    # 给答案流添加watermark
    answersDf = spark.readStream.table(f"{bronze_catalog_name}.{bronze_schema_name}.{bronze_answers_table_name}") \
        .withWatermark("answer_inserted_datetime", "1 hour")

    # 关联时添加时间范围条件：答案插入时间在预约时间之后，且不超过预约时间1天（可根据业务调整）
    df = appointmentsDf.join(
        answersDf,
        [
            appointmentsDf["id"] == answersDf["appointmentId"],
            answersDf["answer_inserted_datetime"] >= appointmentsDf["inserted_datetime"],
            answersDf["answer_inserted_datetime"] <= F.date_add(appointmentsDf["inserted_datetime"], 1)
        ],
        "inner"
    )

    # 生成统一事件时间（取两个时间的最大值）并添加watermark
    df = df.withColumn("event_time", F.greatest("inserted_datetime", "answer_inserted_datetime")) \
        .withWatermark("event_time", "1 hour")

    # 先按appointmentId和event_time聚合，再取每个appointmentId的最新结果
    df = df.groupBy('appointmentId', 'event_time') \
        .agg(F.max(F.when(F.col("prompt") == F.lit(QUESTION_BALL_FITTING), 'answer')).alias('answer')) \
        .groupBy('appointmentId') \
        .agg(F.max(F.struct("event_time", "answer")).alias('latest')) \
        .select('appointmentId', 'latest.answer')

关键说明 #

• 流join时的时间范围条件是核心：让Spark明确超过watermark时间后，不会再有匹配的关联数据，从而可以清理旧的关联状态
• 基于统一event_time的watermark，让Spark能够判断某条appointmentId的聚合结果不会再被后续数据更新，从而支持Append模式输出

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