Spark Structured Streaming流表关联实现
在Spark Structured Streaming中如何实现流表关联（Stream-Stream Join、Stream-Static Join、Stream-Batch Join）？请分别说明三种关联方式的实现原理、状态管理机制和限制条件。如何处理流表关联中的延迟数据问题？
回答
古法程序员
1. Stream-Static Join（流表关联静态表）
val staticDF = spark.read.parquet("/dim/user")
val streamDF = spark.readStream...
streamDF.join(staticDF, "user_id")
- 原理：静态表广播到所有Executor
- 限制：静态表需能全部装入内存
- 更新：静态表不支持热更新（需重建StreamingQuery）
2. Stream-Stream Join（流表关联流表）
stream1.withWatermark("event_time", "10 min")
.join(
stream2.withWatermark("event_time", "5 min"),
Seq("order_id"),
"inner"
)
原理：
- 左右流各自维护状态（State Store）
- 每个流配置Watermark控制状态保留时间
- 迟到的数据超过Watermark后被丢弃
关键约束：
- 两个流都需指定Watermark
- 关联条件必须包含Watermark列的时间范围约束
- 例如：
left_time BETWEEN right_time AND right_time + interval 5 min
3. Stream-Batch Join
val batchDF = spark.read.format("delta").table("orders")
streamDF.join(batchDF, "key") // 转为ForeachBatch模式
- 原理：每个微批次触发时读取Batch表最新快照
- 适用：维度表每天更新一次的场景
4. 延迟数据处理
// Watermark + 状态保留时间
.withWatermark("ts", "1 hour")
// 使用Append模式，聚合结果输出
// Update模式输出更新
5. 状态管理
- 状态存储在State Store（HDFS后端的RocksDB）
- 通过Watermark自动清理过期状态
- 可配置
spark.sql.streaming.stateStore.stateCompressionEnabled
6. 最佳实践
- 优先使用Append模式
- 合理设置Watermark（延迟数据容忍度）
- 对Stream-Stream Join设置较小的状态保留时间
- 使用支持更新的数据源（如Delta Lake）做Stream-Static Join