如何在Structured Streaming中拆分流且避免重复读取Kafka？

在Structured Streaming中，你可以通过缓存从Kafka读取后的原始DataFrame，再基于缓存后的数据集拆分不同事件流，这样就能彻底避免重复拉取Kafka数据的问题，同时不需要额外创建中间主题。

具体实现步骤 #

1. 读取并解析Kafka原始流
   先从全局主题读取数据，完成JSON解析等通用预处理，得到包含所有事件的DataFrame。

   // 以Scala为例，Python逻辑类似
   import org.apache.spark.sql.functions._
   import org.apache.spark.sql.types._
   
   // 定义事件结构
   val eventSchema = StructType(Seq(
     StructField("eventId", StringType),
     StructField("eventType", StringType),
     StructField("eventTime", TimestampType),
     StructField("dataA", StringType),
     StructField("dataB", StringType)
   ))
   
   // 读取Kafka流并解析JSON
   val rawEventDF = spark.readStream
     .format("kafka")
     .option("kafka.bootstrap.servers", "your-kafka-host:9092")
     .option("subscribe", "global-topic")
     .load()
     .selectExpr("CAST(value AS STRING) as json_str")
     .select(from_json(col("json_str"), eventSchema).as("event"))
     .select("event.*")
   

2. 缓存预处理后的DataFrame
   对上述原始事件DataFrame调用cache()或persist()，让Spark复用这部分数据，避免后续拆分流时重复读取Kafka：

   // 缓存数据，默认采用MEMORY_AND_DISK级别，可根据数据量调整存储级别
   val cachedEventDF = rawEventDF.cache()
   

3. 拆分事件流并添加Watermark
   基于缓存后的DataFrame，按eventType过滤得到两个子流，同时为每个流添加Watermark（用于清理过期状态，避免内存溢出）：

   // 拆分类型A的事件流
   val streamA = cachedEventDF
     .filter(col("eventType") === "TYPE_A")
     .withWatermark("eventTime", "10 minutes") // 根据业务场景调整窗口时长
     .select("eventId", "eventTime", "dataA")
   
   // 拆分类型B的事件流
   val streamB = cachedEventDF
     .filter(col("eventType") === "TYPE_B")
     .withWatermark("eventTime", "10 minutes")
     .select("eventId", "eventTime", "dataB")
   

4. 执行流与流的外连接
   基于两个子流执行开窗外连接，注意关联条件需包含时间范围，配合Watermark清理无效状态：

   val joinedStream = streamA.join(
     streamB,
     expr("""
       streamA.eventId = streamB.eventId AND
       streamB.eventTime BETWEEN streamA.eventTime AND streamA.eventTime + INTERVAL 5 minutes
     """),
     joinType = "outer"
   )
   

5. 启动流查询并设置检查点
   最后将连接结果输出到目标存储，同时必须设置检查点路径保证容错：

   joinedStream.writeStream
     .format("parquet") // 根据业务需求选择输出格式（如console、kafka、jdbc等）
     .option("checkpointLocation", "/path/to/your/checkpoint")
     .option("path", "/path/to/output")
     .start()
     .awaitTermination()
   

关键注意事项 #

• 缓存级别选择：如果数据量较大，可改用persist(StorageLevel.MEMORY_ONLY_SER)减少内存占用，或MEMORY_AND_DISK_SER保证数据不丢失。
• Watermark对齐：两个子流的Watermark时长建议保持一致，且连接的时间范围不能超过Watermark的时长，否则无法有效清理状态。
• 检查点必须设置：Structured Streaming依赖检查点恢复状态，包括缓存的数据集状态，重启应用后能从断点继续处理。
• 避免冗余操作：所有通用预处理（如JSON解析、字段提取）都要放在缓存前完成，确保缓存的是最精简的有效数据。

这种方案直接复用了原始Kafka读取的结果，完全省去了中间Kafka主题的读写开销，性能比拆分查询的方案更优，同时逻辑更紧凑。

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