Spark Structured Streaming流静态DF按条件关联写Kafka报错求解

• 你当前的实现将同一份流DataFrame streamDF 做了两次过滤拆分，生成df_join1和df_join2后再做union，Spark Structured Streaming在生成物理执行计划时，会把这两次对streamDF的引用识别为两个独立的流数据源，所以整个查询的数据源数量变为2个。
• 你使用的旧checkpoint目录中存储的是之前只有1个流源的查询的偏移量、拓扑元数据，重启查询时Spark会校验新旧查询的拓扑一致性，发现源数量不匹配就会抛出该断言错误。
• 额外的副作用：这种写法会导致同一份流数据被重复消费两次，浪费集群资源。

1. 首先调整关联逻辑，避免拆分流再union，用多条件分支合并关联逻辑，全链路只保留1个流源：

// 静态DF字段加前缀避免和流DF字段冲突，可根据实际业务调整
val prefixedStaticDF = staticDF.select(staticDF.columns.map(colName => col(colName).as(s"static_${colName}")):_*)

val resultDF = streamDF.as("a")
  .join(
    prefixedStaticDF.as("b"),
    // 合并不同col1取值对应的关联条件
    (col("a.col1").isin("value1", "value2") && joinCondition01) ||
    (col("a.col1").isin("value3", "value4") && joinCondition02),
    "left_outer"
  )

如果静态DF数据量较小，建议用广播变量优化关联性能，避免shuffle，写法如下：

import org.apache.spark.sql.functions.broadcast

val resultDF = streamDF.as("a")
  .join(
    broadcast(prefixedStaticDF).as("b"),
    (col("a.col1").isin("value1", "value2") && joinCondition01) ||
    (col("a.col1").isin("value3", "value4") && joinCondition02),
    "left_outer"
  )

2. 处理checkpoint问题：如果你的业务允许丢数或者重跑历史数据，直接删除旧的两个checkpoint目录/checkpointLocation/tmp_01、/checkpointLocation/tmp_02后重启即可；如果不允许重跑，需要保证新的查询逻辑和checkpoint对应的旧逻辑拓扑完全一致，或者切换新的checkpoint目录后手动指定起始消费偏移量。
3. 你的多输出逻辑（同一份结果写入两个Kafka Topic）的写法本身是合规的，不需要调整，只要保证每个流查询对应独立的checkpoint目录即可。

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