Spark流-流连接返回空结果，求原因排查及解决方法

核心原因分析 #

1. 时间匹配条件逻辑错误
   你的连接条件中ts_a >= ts_b AND ts_a <= ts_b等价于强制要求ts_a = ts_b，但Table A的时间是2021年12月，Table B是2022年5月，两者时间差6个月，完全没有满足ts_a = ts_b的记录，这是导致空结果的直接原因。

2. 水印配置与业务时间范围不匹配
   你给两个流都设置了1分钟的水印（withWatermark('ts', '1 minute')），意味着Spark会清理掉超过当前处理时间1分钟的旧数据。Table A的时间远早于Table B，当处理Table B的数据时，Table A的对应数据已经被水印规则清理，即使调整时间条件，也无法完成匹配。

3. Outer Join的流处理约束
   Spark Streaming的Outer Join要求连接条件必须包含基于水印的时间范围约束，否则Spark无法确定何时可以安全输出无匹配的单边数据，可能会延迟输出或直接丢弃数据。

解决方案 #

1. 修正连接时间条件 #

根据实际业务需求调整时间匹配规则：

• 如果仅需要按ID和Ref关联（不限制时间），可以直接删除时间相关条件：

  cond = expr("""
  ID_a = ID_b AND
  Ref_a = Ref_b
  """)
  

• 如果需要在一定时间范围内关联，比如允许A和B的时间差在6个月内，调整时间条件：

  cond = expr("""
  ID_a = ID_b AND
  Ref_a = Ref_b AND
  ts_a >= ts_b - interval '6 months' AND
  ts_a <= ts_b + interval '6 months'
  """)
  

2. 调整水印时长 #

针对时间范围较大的关联，需要设置足够长的水印来保留历史数据，确保匹配时数据未被清理：

df_a = spark.readStream.schema(...).parquet("s3:...").withWatermark('ts_a', '7 months')
df_b = spark.readStream.schema(...).parquet("s3:...").withWatermark('ts_b', '7 months')

3. 验证Schema正确性 #

确保读取流时定义的Schema中，ts_a和ts_b的类型是TimestampType，避免因类型不匹配导致的关联失败：

from pyspark.sql.types import StructType, StructField, IntegerType, StringType, TimestampType

schema_a = StructType([
    StructField("ID_a", IntegerType()),
    StructField("Ref_a", StringType()),
    StructField("ts_a", TimestampType())
])

4. 测试小批量数据 #

可以先使用静态DataFrame测试连接逻辑，确认条件正确后再切换到流处理：

# 静态测试
df_a_static = spark.read.schema(schema_a).parquet("s3:...")
df_b_static = spark.read.schema(schema_b).parquet("s3:...")
joined_static = df_a_static.join(df_b_static, cond, 'outer')
joined_static.show()

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