📚 课程目录第3章 Structured Streaming核心概念

3. Structured Streaming 核心概念：无界表、Event-Time 与 Watermark、输出模式

在信用额度定价的流式场景中，交易事件源源不断地到达，系统需要像处理数据库表一样处理这些“无界”的数据流。Structured Streaming 正是基于“无界表”（Unbounded Table）这一核心抽象，将批处理引擎的成熟语义无缝扩展到流处理领域。理解无界表、Event-Time 与 Watermark、以及三种输出模式，是构建可靠、低延迟的信用额度决策管道的基础。

3.1 无界表：流数据的核心抽象

Structured Streaming 将输入数据流视为一张持续追加的“无界表”。每个到达的数据记录，都相当于向这张表中插入一行新数据。系统会周期性地对这张无界表执行增量查询（类似批处理中的 SELECT），并将结果写入到“结果表”（Result Table）中。

  核心思想： 流计算 = 对无界表的连续微批（Micro-Batch）或连续处理（Continuous Processing）查询。

Append（追加）：仅将结果表中新增的行写出。适用于无聚合的简单过滤/映射。 • Update（更新）：仅将结果表中发生更新的行写出。适用于有聚合但允许部分更新的场景。 • Complete（完全）：每次触发都将整个结果表写出。适用于有聚合且结果集较小的场景。

实践要点： 在信用额度定价中，输入无界表通常包含 transaction_id、user_id、amount、timestamp 等字段。Spark 会持续监控数据源（如 Kafka），将新到达的 JSON 或 Avro 记录解析为 DataFrame 的行。

3.2 Event-Time 与 Watermark：处理乱序与延迟

在分布式系统中，事件到达 Spark 引擎的时间（Processing Time）与事件实际发生的时间（Event-Time）往往不一致。网络延迟、背压、重试等因素会导致数据乱序到达。为了在“事件发生时间”维度上进行有意义的聚合（如“最近 5 分钟每个用户的累计交易额”），必须使用 Event-Time，并引入 Watermark 来容忍延迟。

3.2.1 Event-Time 与 Processing-Time 对比

维度	Event-Time（事件时间）	Processing-Time（处理时间）
定义	事件实际发生的时间戳（由数据源生成）	事件到达 Spark 引擎的系统时间
可靠性	反映业务真实时序，不受系统负载影响	受系统负载、重试、调度延迟影响
适用场景	窗口聚合、累计额度计算、异常检测	简单监控、无状态过滤、调试
乱序容忍	需要 Watermark 机制配合	天然按到达顺序处理，但结果不准确

3.2.2 Watermark：延迟容忍的边界

Watermark 是一个时间阈值，表示“系统认为在此时间之前的事件已经全部到达”。Spark 会持续跟踪当前已观察到的最大的 Event-Time，并减去用户指定的延迟容忍度（例如 10 分钟）。任何 Event-Time 小于当前 Watermark 的迟到数据将被丢弃或忽略，从而保证窗口聚合的最终性。

  Watermark 公式： Watermark = max(observed_event_time) - delayThreshold

  示例： 若最大事件时间为 12:30，延迟容忍度为 10 分钟，则 Watermark = 12:20。任何事件时间早于 12:20 的记录将被视为“太迟”而丢弃。

注意： Watermark 设置过小会导致大量迟到数据被丢弃，造成聚合结果不准确；设置过大会增加状态存储压力，延迟窗口关闭时间。在信用额度定价中，建议根据网络延迟和业务容忍度（通常 5-15 分钟）进行调优。

3.3 三种输出模式：Append / Update / Complete

输出模式决定了每次触发（Trigger）后，如何将结果表的变化写入外部存储（如 Kafka、HDFS、JDBC）。选择正确的输出模式，直接影响系统的语义正确性和性能。

3.3.1 模式对比与选择

输出模式	行为描述	适用场景	状态要求	典型示例
Append	仅将结果表中新增的行写出，已有行不修改	无聚合的过滤、映射、投影；或使用 Watermark 的聚合	无状态或仅追加	实时交易日志归档
Update	仅将结果表中发生更新的行写出（新增或修改）	有聚合且结果集持续更新，但不需要全量输出	有状态（Keyed State）	实时用户额度累计
Complete	每次触发都将整个结果表全部写出	有聚合且结果集较小（如全局统计、Top-N）	有状态（全量）	全局交易总额统计

3.3.2 代码示例：三种模式在额度定价中的应用

// 假设输入流包含交易事件
val inputDF = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "localhost:9092")
  .option("subscribe", "transactions")
  .load()
  .selectExpr("CAST(value AS STRING) as json")
  .select(from_json($"json", schema).as("data"))
  .select("data.user_id", "data.amount", "data.event_time")

// 1. Append 模式：简单过滤高风险交易
val highRiskDF = inputDF.filter($"amount" > 10000)
val appendQuery = highRiskDF.writeStream
  .outputMode("append")
  .format("console")
  .start()

// 2. Update 模式：按用户累计额度（带 Watermark）
val userLimitDF = inputDF
  .withWatermark("event_time", "10 minutes")
  .groupBy($"user_id", window($"event_time", "5 minutes"))
  .agg(sum($"amount").as("total_amount"))
val updateQuery = userLimitDF.writeStream
  .outputMode("update")
  .format("console")
  .start()

// 3. Complete 模式：全局每分钟交易总额
val globalStatsDF = inputDF
  .withWatermark("event_time", "10 minutes")
  .groupBy(window($"event_time", "1 minute"))
  .agg(sum($"amount").as("global_total"))
val completeQuery = globalStatsDF.writeStream
  .outputMode("complete")
  .format("console")
  .start()

选型建议： 在信用额度定价中，Update 模式是最常用的选择。因为每个用户的额度是持续更新的键值对，使用 Update 模式可以只输出发生变化的用户额度，大幅减少下游存储的写入压力。Append 模式适合日志归档，Complete 模式适合全局监控面板。

3.4 本章小结

无界表是 Structured Streaming 的基石，将流数据抽象为持续追加的表，支持 SQL 语义的增量查询。
Event-Time 是业务真实时间，Watermark 是容忍乱序的边界，两者结合才能保证窗口聚合的准确性和最终性。
三种输出模式（Append / Update / Complete）决定了结果表的写出策略，需根据聚合类型和结果集大小合理选择。
在信用额度定价场景中，推荐使用 Event-Time + Watermark + Update 模式 的组合，实现低延迟、高准确度的流式额度计算。