Spark Structured Streaming Kafka流左外自连接水位线报错求助

问题背景与操作代码 #

尝试在Spark Structured Streaming中执行流-流左外自连接，用于后续拆分已连接和未连接的行，操作代码如下：

读取Kafka流数据 #

df_source = app.spark.readStream.format("kafka") \
                 .option("kafka.bootstrap.servers", "XXXXXXX") \
                 .option("subscribe", "XXXXXXX") \
                 .option("startingOffsets", "earliest") \
                 .option("kafka.group.id", "924ee006-c268-11ed-afa1-0242ac120002") \
                 .load() 

df_source.createOrReplaceTempView("df_source")

解析JSON数据 #

df0 = app.spark.sql("""\
SELECT
    CAST(value AS STRING) AS json
FROM df_source
""")

df0.createOrReplaceTempView("df0")

df1 = app.spark.sql("""\
SELECT
    from_json(json,        'struct< `metadata`   : struct< `namespace` : string 
                                                         , `name`      : string
                                                         , `name0`     : string 
                                                         , `size0`     : int
                                                         , `message0`  : struct< `id`           : string
                                                                               , `type`         : string
                                                                               , `timestamp`    : string
                                                                               , `date`         : string
                                                                               , `time`         : string
                                                                               , `process_name` : string
                                                                               , `loglevel`     : string
                                                                               , `process_id`   : string
                                                                               >
                                                         >
                                  , `spec`       : struct< `fix`                  : string
                                                         , `source_process_name`  : string
                                                         , `sink_process_name`    : string
                                                         , `source_CLORDID`       : string
                                                         , `sink_CLORDID`         : string
                                                         , `action`               : string
                                  >
                                  , `@timestamp` : timestamp
                                  >
                           ')               AS dict
FROM df0
""")

df1.createOrReplaceTempView("df1")

定义水位线并执行自连接 #

df1_new = df1.withWatermark("`dict.@timestamp`", "2 minutes")

df2 = df1.alias("orig").join(df1_new.alias("new"), 
                     expr("""
                         orig.dict.spec.sink_CLORDID = new.dict.spec.source_CLORDID AND
                         \"new.dict.@timestamp\" >= \"orig.dict.@timestamp\" AND
                         \"new.dict.@timestamp\" <= \"orig.dict.@timestamp\" + interval 1 minute
                    """),
                     "leftOuter"
                )

报错信息 #

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. 逐步从流定义待连接DataFrame，是否需为所有DataFrame定义水位线以清理状态？

问题解答 #

1. 报错原因与修正方案 #

你的代码存在两个核心问题导致水位线校验失败：

• 水位线仅配置在单侧DataFrame：左外连接中，左表orig未设置水位线，Spark要求流-流左外连接必须满足：要么连接键包含水位线字段，要么nullable侧（右表new）设水位线且左表有对应时间范围约束，而你的左表既无水位线，时间条件的写法也错误。
• 时间条件的引号使用错误：expr中用转义双引号包裹字段名，导致Spark将其识别为字符串常量而非字段，无法构建有效的时间范围约束。

修正后的代码：

# 给用于连接的DataFrame统一设置水位线
df1_with_watermark = df1.withWatermark("`dict.@timestamp`", "2 minutes")

df2 = df1_with_watermark.alias("orig").join(df1_with_watermark.alias("new"), 
                     expr("""
                         orig.dict.spec.sink_CLORDID = new.dict.spec.source_CLORDID AND
                         new.`dict.@timestamp` >= orig.`dict.@timestamp` AND
                         new.`dict.@timestamp` <= orig.`dict.@timestamp` + interval 1 minute
                    """),
                     "leftOuter"
                )

自连接场景下，给两侧流都设置水位线，能让Spark基于水位线清理过期连接状态，避免状态无限膨胀。

2. 水位线的设置范围 #

不需要给所有中间DataFrame设置水位线，仅需在最终用于连接的DataFrame上配置即可。中间的df_source、df0属于转换步骤，只要最终的df1（或其带水位线的版本）基于流数据生成且配置正确水位线，就能满足状态清理需求。

注意：水位线绑定在DataFrame的事件时间字段上，设置后后续转换会继承该配置，因此只需在连接前的最后一步给目标DataFrame添加水位线即可。

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