4、Structured Streaming 基础：流式数据源、输出模式与触发器

好，咱们进入实时入湖的第一道门槛——Structured Streaming。说实话，很多同学一开始就被「流式」两个字吓住了，觉得跟批处理是两套完全不同的东西。其实不然。Structured Streaming 的设计哲学就是「把流当成无限的表来操作」。你想想看，批处理是对一张有限表做查询，流处理只不过是把这张表变成了无限追加的数据而已。

我个人习惯把 Structured Streaming 拆成三个核心要素来理解：数据从哪来（Source）、怎么处理（Query）、结果写哪去（Sink）。这一章咱们重点啃前两个——流式数据源和输出模式，触发器是控制节奏的关键，也得一并拿下。

4.1 流式数据源：实时数据的入口

Structured Streaming 内置了几种常用的数据源，我挑最常用的三个讲。

File Source 是我最早接触的流式源。它本质上就是不断扫描一个目录，把新出现的文件当作新批次读进来。我在项目中遇到过一个问题：如果文件写入不是原子操作（比如正在写就被 Spark 扫描到了），会读到不完整的数据。后来我强制要求上游用 .copy 或 .tmp 后缀写完再重命名，才彻底解决。

// 监控 HDFS 上的日志目录
val df = spark.readStream
  .format("json")
  .schema(logSchema)
  .option("maxFilesPerTrigger", 10)   // 每批次最多处理10个文件
  .load("/data/logs/")

Kafka Source 是生产环境的主力。它的核心是 offset 管理——你消费到哪条消息，Spark 会记录在 checkpoint 里。我曾经帮一个团队排查过数据重复问题，最后发现是他们把 startingOffsets 设成了 "earliest" 但 checkpoint 目录没配对，每次重启都从头消费。嗯，这里要注意：checkpoint 是 Exactly-Once 的命根子。

val kafkaDF = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "host1:9092,host2:9092")
  .option("subscribe", "topic_orders")
  .option("startingOffsets", "latest")   // 生产环境建议用 latest
  .load()

4.2 输出模式：决定你看到什么

输出模式说白了就是——每次触发计算后，怎么把结果写出去。有三种模式，我按使用频率排个序。

Append 模式 最直观。你做一个 filter 或 select，来一条数据就输出一条。我做的第一个实时入湖方案用的就是 Append——从 Kafka 读 JSON，简单清洗后直接写 Delta 表。这种模式对状态管理要求最低，性能也最好。

Update 模式 适合做聚合。比如统计每个用户的订单数，新订单来了，只输出该用户更新后的计数。我遇到过一个问题：如果聚合的 key 基数特别大（比如上亿用户），Update 模式的状态存储会膨胀得很厉害。后来我加了 withWatermark 来清理过期状态，才稳住。

val aggDF = kafkaDF
  .groupBy("user_id")
  .count()
  .writeStream
  .outputMode("update")
  .format("console")
  .start()

Complete 模式 每次把整个结果集写一遍。你想想看，如果聚合结果有 100 万行，每次触发都写 100 万行，这谁受得了？所以它只适合结果集很小的场景，比如全局 PV 计数。

4.3 触发器：控制处理节奏

触发器决定了「多久触发一次计算」。Structured Streaming 提供了三种触发策略。

默认微批 是最稳的。它像是一个勤劳的工人，干完活马上问「还有吗？」。延迟通常在秒级，适合大部分实时入湖场景。

固定间隔 我用的不多。因为它有个问题：如果处理时间超过了间隔，下一批会排队，造成背压。我曾经在一个项目中为了控制写入 Delta 的频率，用了 30 秒的固定间隔，结果数据积压越来越严重。后来改成默认微批 + 限流 maxOffsetsPerTrigger，反而更平滑。

df.writeStream
  .trigger(ProcessingTime("10 seconds"))
  .format("delta")
  .start("/path/delta-table")

一次性触发器 是个宝贝。它让流式查询只跑一次就结束，本质上变成了批处理。我经常用它来做「增量 ETL」——每天凌晨用 Once() 把前一天积累的流式数据一次性处理完，既享受流式 API 的便利，又不用长期占用资源。

连续处理 是 Spark 2.3 引入的实验特性，延迟能压到毫秒级。但限制很多：不支持 sort、不支持某些聚合、容错机制也不如微批成熟。我建议你先观望，等它 GA 了再考虑。

4.4 知识体系总览

下面这张图把 Structured Streaming 的核心脉络串起来了。从数据源进来，经过输出模式的选择，再由触发器控制节奏，最终写出到 Sink。我建议你把它当作一张地图，遇到问题时先定位是哪个环节出了岔子。

4.5 实战组合：一个完整的流式读取 + 输出示例

把上面三个要素串起来，就是一个完整的流式作业。下面这个例子从 Kafka 读取订单数据，做简单的过滤，然后以 Append 模式写入 Delta 表，每 10 秒触发一次。

import org.apache.spark.sql.streaming.Trigger

val ordersDF = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "localhost:9092")
  .option("subscribe", "orders")
  .option("startingOffsets", "latest")
  .load()
  .selectExpr("CAST(value AS STRING) as json")
  .select(from_json($"json", orderSchema).as("data"))
  .select("data.*")

val validOrders = ordersDF.filter($"amount" > 0)

validOrders.writeStream
  .format("delta")
  .outputMode("append")
  .trigger(Trigger.ProcessingTime("10 seconds"))
  .option("checkpointLocation", "/tmp/checkpoint/orders")
  .start("/data/delta/orders")