§ 01 · 设计思想

为什么要实时计算？批处理不够用在哪里

在电商秒杀、风控检测、物联网告警这些场景里，数据的价值随时间急速衰减——一笔欺诈交易如果 10 分钟后才被发现，损失已经无法挽回。批处理（Batch Processing）每隔几小时跑一次 MapReduce 或 Spark Job，就像每晚才看一次当天的监控录像，对“当下正在发生的事”完全无能为力。

实时计算要解决的核心矛盾是：数据是连续到来的无限流，而计算资源是有限的。Kafka 可以缓冲流数据，但它不会帮你聚合、关联、窗口化。你需要一个流处理引擎。

Spark 选择了一条与 Flink 不同的路：微批（Micro-Batch）哲学。与其把每条记录当作真正的流，不如把流切成极小的批，用成熟的批引擎高速处理每一个小批次。这个决定换来了工程上的巨大红利：容错、SQL 支持、与批处理代码的统一——代价是延迟比真正的流引擎高出一个数量级（百毫秒级 vs 毫秒级）。

核心结论
Spark 的微批本质是“用可接受的延迟换取工程简洁性”。 它让你用几乎相同的 DataFrame API 同时写批处理和流处理代码，这对团队协作是巨大优势。
如果你的业务对延迟要求是 秒级以内，Spark Structured Streaming 完全够用；如果需要 毫秒级（如高频交易），请考虑 Apache Flink。

实战口诀：“Spark 流是高速批，批的快就是流的慢；延迟要求秒以内，结构化流够用了。”

---

§ 02 · 整体架构

Structured Streaming 顶层全景

Spark 的实时计算体系经历了两代：第一代 DStream API（Spark 1.x）是 RDD 的流式包装；第二代 Structured Streaming（Spark 2.0+）基于 DataFrame/Dataset，被官方定为长期维护的主流 API。本章以 Structured Streaming 为核心，DStream 作为对比了解。

下图展示了 Structured Streaming 的完整数据流：数据从外部 Source 进入，经过 Query Plan 处理，写出到 Sink，同时 Checkpoint 机制保证容错。

核心角色一览

StreamExecution 是整个引擎的大脑，负责维护 Trigger 调度循环。每次 Trigger 触发时，它从 Source 读取新 offsets，构建本次微批的 Logical Plan，交给 Catalyst 优化成 Physical Plan，最终作为普通 Spark Job 提交执行。

State Store 是有状态算子（如 groupBy + count）的内存+磁盘存储，基于 RocksDB（Spark 3.2+）或内置的 HDFS-backed 实现，保证 Executor 重启后状态不丢失。

Watermark 是处理迟到数据的时间边界。流数据中事件的“产生时间”和“到达时间”往往不一致（网络延迟、设备离线），Watermark 告诉引擎：“比当前水位线早 X 分钟以前的数据，直接丢弃，不再等待”。

核心结论
Structured Streaming 本质上是一个“增量式批处理器”：每次 Trigger 产生一个微小的 Spark Job，处理从上次偏移量到当前偏移量之间的新数据，然后把结果更新到 Sink，把状态更新到 State Store，把进度更新到 Checkpoint。
理解这一点，所有的配置问题（为什么延迟高？为什么 State 越来越大？为什么重启后重复消费？）都有了答案。

实战口诀：“Streaming = 高速循环的 Batch；Source 提供偏移量，Checkpoint 记住偏移量，State Store 记住聚合值。”

---

§ 03 · 关键机制一：微批执行循环

Trigger 与微批调度：引擎的心跳

StreamExecution 的工作方式就像一个永不停歇的心跳循环。每一次心跳（Trigger）完成以下四个步骤：

四种 Trigger 类型

Spark 提供四种 Trigger 策略，决定了“多久产生一次微批”：

类型	说明
ProcessingTime(“2 seconds”)	每 2 秒触发一次微批，如果上次批次未完成则跳过等待。适合大多数生产场景，延迟 = trigger间隔 + 处理时间。
Continuous(“100 milliseconds”)	真流模式，延迟低至 ~1ms，仅支持简单 map/filter 变换，不支持聚合、窗口。实验性功能，生产慎用。
Once()	执行一次微批后退出，用于增量式定时任务。Spark 3.3+ 推荐用 AvailableNow。
AvailableNow()	处理当前所有可用数据后退出，可分多批次处理，比 Once 更可靠，适合 Lambda 架构的增量批。

核心结论
Trigger 决定了吞吐和延迟的权衡。 间隔越短延迟越低，但每次批次的数据量越少、调度开销占比越高；间隔越长吞吐越好，但延迟越高。生产中 ProcessingTime("10 seconds") 到 ProcessingTime("1 minute") 最为常见。
一个关键的认知：如果当前批次处理时间超过了 Trigger 间隔，Spark 不会同时跑两个批次——下一次 Trigger 会等待当前批次结束再开始，保证批次间有序性。

实战口诀：“Trigger 是节拍器，ProcessingTime 最常用；批次超时不并发，串行保证有序性。”

---

§ 04 · 关键机制二：Watermark 与迟到数据

水位线：处理“迟到的世界”

流数据最令人头疼的问题之一是乱序（Out-of-order）。想象手机 App 在地铁断网时积攒了 5 分钟的行为日志，一旦重新联网就批量上传——这些数据的“事件时间”（Event Time）在 5 分钟前，但“到达时间”（Processing Time）是现在。

如果你在做每分钟的点击量统计，这批数据应该归到 5 分钟前的窗口，但那个窗口可能已经关闭并输出了。Watermark 就是解决这个问题的机制：它是引擎对“我愿意等多久”的承诺。

Watermark 与窗口聚合的联动

Watermark 的触发逻辑是：当水位线推进到某个窗口的结束时间时，该窗口被“关闭”并输出结果。具体来说：

• 📌 withWatermark("eventTime", "2 minutes")：声明最多容忍 2 分钟的迟到
• 📌 水位线 = 已观测到的最大 event_time - 2 minutes
• 📌 当水位线 > 窗口结束时间时，该窗口在 Append 模式下才会输出
• 📌 event_time < 水位线的记录：直接丢弃，不参与任何窗口计算

常见陷阱： 在 Update 或 Complete 输出模式下，Watermark 不会触发状态清理，所有历史窗口都保留在 State Store 中，会导致状态无限增长！只有在 Append 模式 下，Watermark 才会触发窗口关闭并清理状态。

实战口诀：“水位线是引擎的宽容度；Append 模式才触发状态清理，Update/Complete 模式慎用有状态聚合。”

---

§ 05 · 关键机制三：三种 Output Mode

输出模式：决定“把什么写出去”

Structured Streaming 有三种输出模式，它们的区别在于“每次 Trigger 时，把结果表的哪些行写到 Sink”。选错输出模式是初学者最常遇到的报错之一。

Update 模式

Update 模式：每次只输出本批次中发生变化的行（既不是全量，也不是只新增行）。适合“键值更新”场景，如每隔一段时间更新某个 key 的计数。但注意 Update 模式下 Watermark 不会清理历史 key 的状态，需要配合 mapGroupsWithState 手动管理超时清理。

三种模式的适用场景总结：

• Append 无聚合的转换、配合 Watermark 的窗口聚合 → 写 Kafka / 文件
• Complete 全局聚合（实时排行榜、全局统计）→ 写数据库（覆盖）
• Update 有状态的 key 级别更新 → 写支持 upsert 的数据库（如 HBase、Cassandra）

实战口诀：“Append 追加只写新，Complete 全量覆盖写，Update 只写改变行；有聚合不加 Watermark，不能用 Append。”

---

§ 06 · 关键机制四：State Store 状态管理

状态存储：流处理的“工作内存”

流处理与批处理最大的区别在于：批处理每次都从头计算，流处理需要在微批之间记住中间结果。这个“记住”的能力，就是 State Store。

类比：把 State Store 想象成一个跨批次共享的 HashMap——每个 key 对应一个聚合值，每次微批更新这个 HashMap，下一批读取上一批留下的值继续累加。

启用 RocksDB State Store

生产环境中，当有状态聚合的 key 数量超过百万时，内置的基于堆内存的 State Store 会导致严重的 GC 停顿。RocksDB State Store 将状态持久化到 Executor 的本地磁盘，只有热数据缓存在内存中，是处理大状态的首选。

RocksDB 启用条件： Spark 3.2+，在 SparkConf 中设置 spark.sql.streaming.stateStore.providerClass=org.apache.spark.sql.execution.streaming.state.RocksDBStateStoreProvider。每个 Executor 需要有足够的本地磁盘空间存储状态快照。

实战口诀：“状态是流的记忆；key 少用内存存，key 多用 RocksDB；Complete 模式 + 无 Watermark = 状态无限增长。”

---

§ 07 · 新老对比

DStream vs Structured Streaming：两代 API 的本质区别

Spark 历史上有两套流处理 API，了解它们的差异有助于维护老代码，也能理解 Structured Streaming 为什么是“正确”的设计。

特性	DStream（第一代）	Structured Streaming（第二代）
基础	基于 RDD，操作是函数式的	基于 DataFrame/Dataset，统一 API
时间语义	只有 Processing Time	支持 Event Time、Processing Time
乱序处理	无内置 Watermark 支持	内置 Watermark 处理乱序数据
窗口操作	API 独立于批处理	与 groupBy 统一语法
Checkpoint	需手动管理 checkpoint 位置	自动管理 offsets checkpoint
SQL 支持	无	完整 SQL 支持
生态	与 Spark SQL 生态割裂	与 Delta Lake、MLlib 深度集成
代码复用	Streaming 和 Batch 代码不能复用	Batch/Stream 代码高度复用

新项目请不要使用 DStream！ DStream API 在 Spark 3.x 已进入维护模式，官方不会再为其添加新功能。Structured Streaming 是官方长期支持的流处理 API，生态也更完整（Delta Lake 的流读写仅支持 Structured Streaming）。

---

§ 08 · 生产实战

完整生产级代码示例

下面是一个真实的生产场景：从 Kafka 读取用户行为事件（JSON 格式），做基于事件时间的每分钟窗口聚合，统计每个页面的 PV，然后写出到 Kafka 和文件。包含完整的 Watermark、Checkpoint、RocksDB 配置。

# =============================================================
# 生产级 Structured Streaming 示例
# 场景：Kafka → 每分钟窗口 PV 统计 → Kafka + Parquet
# =============================================================

from pyspark.sql import SparkSession
from pyspark.sql.functions import (
    col, from_json, window, count, current_timestamp
)
from pyspark.sql.types import (
    StructType, StructField, StringType, LongType
)

# ── 1. SparkSession 配置 ──────────────────────────────────────
spark = (
    SparkSession.builder
    .appName("realtime_pv_job")
    # 【关键配置】启用 RocksDB State Store，支持超大状态
    .config(
        "spark.sql.streaming.stateStore.providerClass",
        "org.apache.spark.sql.execution.streaming.state"
        ".RocksDBStateStoreProvider"
    )
    # 【关键配置】shuffle partition 数量：流处理建议设为 Executor 核数的倍数
    .config("spark.sql.shuffle.partitions", "20")
    # 【常见错误】不要用默认值 200！流处理中 200 个 shuffle partition
    #   会产生大量小任务，严重影响性能
    .getOrCreate()
)
spark.sparkContext.setLogLevel("WARN")

# ── 2. 定义 Kafka Source ──────────────────────────────────────
# 事件 Schema: {"user_id":"u1","page":"/home","ts":1700000000000}
event_schema = StructType([
    StructField("user_id", StringType(), True),
    StructField("page",    StringType(), True),
    StructField("ts",      LongType(),   True),  # 毫秒时间戳
])

raw_stream = (
    spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", "kafka1:9092,kafka2:9092")
    .option("subscribe", "user_events")
    # 【关键配置】每批次最大拉取量，防止单批次过大导致 OOM
    .option("maxOffsetsPerTrigger", 100000)
    # 【关键配置】earliest = 从最早的消息开始；latest = 只处理新消息
    # 生产中第一次运行用 earliest，之后靠 checkpoint 恢复
    .option("startingOffsets", "earliest")
    .load()
)

# ── 3. 解析 JSON ──────────────────────────────────────────────
# Kafka 的 value 字段是 binary，需要先转 string 再解析 JSON
parsed = (
    raw_stream
    .select(from_json(col("value").cast("string"), event_schema).alias("e"))
    .select("e.*")
    # 把毫秒时间戳转成 Timestamp 类型（Event Time 的基础）
    .withColumn("event_time", (col("ts") / 1000).cast("timestamp"))
    # 【错误写法】.withColumn("event_time", current_timestamp())
    # 这会用 Processing Time，失去事件时间语义！
)

# ── 4. Watermark + 窗口聚合 ───────────────────────────────────
pv_agg = (
    parsed
    # 【关键】声明 Watermark：最多容忍 2 分钟的迟到
    # 含义：水位线 = max(event_time) - 2min
    # 当水位线 > 窗口结束时间，该窗口才在 Append 模式下输出
    .withWatermark("event_time", "2 minutes")
    # 每 1 分钟的翻滚窗口（tumbling window）
    # 如需滑动窗口：window("event_time", "1 minute", "30 seconds")
    .groupBy(
        window(col("event_time"), "1 minute"),
        col("page")
    )
    .agg(count("*").alias("pv"))
    # 把 window struct 展开为 start / end 两列，方便下游消费
    .select(
        col("window.start").alias("window_start"),
        col("window.end").alias("window_end"),
        col("page"),
        col("pv")
    )
)

# ── 5. 写出到 Kafka ───────────────────────────────────────────
# 【关键】有 Watermark 的窗口聚合必须用 Append 模式
# 【错误写法】outputMode("update") → Watermark 不会清理 State
kafka_query = (
    pv_agg
    .selectExpr(
        "CAST(page AS STRING) AS key",
        "to_json(struct(*)) AS value"
    )
    .writeStream
    .format("kafka")
    .outputMode("append")             # ← 必须是 append
    .option("kafka.bootstrap.servers", "kafka1:9092,kafka2:9092")
    .option("topic", "pv_result")
    # 【关键配置】checkpoint 路径，保证重启后不重复消费
    # 每个 writeStream 必须有独立的 checkpoint 路径！
    .option("checkpointLocation", "s3://my-bucket/checkpoints/pv_kafka")
    .trigger(processingTime="10 seconds")  # 每 10 秒一个微批
    .start()
)

# ── 6. 同时写出到 Parquet（ForeachBatch 自定义写出）────────────
def write_to_parquet(batch_df, batch_id):
    """
    ForeachBatch：把每个微批的 DataFrame 用普通批处理 API 写出
    优点：可以用所有批处理 Sink（JDBC、自定义等）
    注意：batch_df 是普通 DataFrame，不是 Streaming DataFrame
    """
    if batch_df.count() > 0:  # 空批次直接跳过
        (
            batch_df.write
            .mode("append")
            .partitionBy("window_start")
            .parquet("s3://my-bucket/pv_data/")
        )

parquet_query = (
    pv_agg
    .writeStream
    .outputMode("append")
    .option("checkpointLocation", "s3://my-bucket/checkpoints/pv_parquet")
    .trigger(processingTime="10 seconds")
    .foreachBatch(write_to_parquet)
    .start()
)

# ── 7. 等待所有查询完成 ───────────────────────────────────────
# awaitAnyTermination：任何一个 query 失败就退出整个程序
spark.streams.awaitAnyTermination()
# 【错误写法】kafka_query.awaitTermination() 只等待一个 query
# 多个 query 时，应该用 awaitAnyTermination

# ════════════════════════════════════════════════
# ❌ 错误写法 1：聚合后不用 Watermark，却用 Append 模式
# ════════════════════════════════════════════════
df.groupBy("page").count()  # 无 Watermark
  .writeStream
  .outputMode("append")  # ← 报错：AnalysisException
  .start()
# 错误原因：无 Watermark 的聚合，引擎不知道何时能"确定"某个 key 不会再更新
# 因此 Append 模式无法工作

# ✅ 正确写法 1：加 Watermark 或改用 Update/Complete 模式
df.withWatermark("event_time", "5 minutes")
  .groupBy(window("event_time", "1 minute"), "page").count()
  .writeStream
  .outputMode("append")  # ← 正确
  .start()

# ════════════════════════════════════════════════
# ❌ 错误写法 2：两个 Query 共用同一 Checkpoint 路径
# ════════════════════════════════════════════════
query1 = df.writeStream.option("checkpointLocation", "/ckpt/job1").start()
query2 = df.writeStream.option("checkpointLocation", "/ckpt/job1").start()
# 错误原因：两个 Query 会争抢同一 Checkpoint 目录，导致数据错乱

# ✅ 正确写法 2：每个 Query 独立的 Checkpoint 路径
query1 = df.writeStream.option("checkpointLocation", "/ckpt/job1_kafka").start()
query2 = df.writeStream.option("checkpointLocation", "/ckpt/job1_parquet").start()

# ════════════════════════════════════════════════
# ❌ 错误写法 3：修改 Query 后复用旧 Checkpoint
# ════════════════════════════════════════════════
# 修改了 groupBy 的列，但 Checkpoint 目录没变
# 运行时会报 StreamingQueryException：Schema 不兼容
# 或者悄悄产生错误结果！

# ✅ 正确写法 3：修改 Query 逻辑后，删除旧 Checkpoint 重新运行
# （或使用新的 Checkpoint 路径）

# ════════════════════════════════════════════════
# ❌ 错误写法 4：shuffle partition 忘记调小
# ════════════════════════════════════════════════
spark = SparkSession.builder.getOrCreate()
# 默认 spark.sql.shuffle.partitions = 200
# 流处理中每 10 秒一个批次，200 个小任务开销巨大

# ✅ 正确写法 4：根据 Executor 核数设置
spark = (
    SparkSession.builder
    .config("spark.sql.shuffle.partitions", "20")  # 约等于总核数
    .getOrCreate()
)

---

§ 09 · 本章总结

知识地图回顾

全章三大核心认知

1. 📌Spark 流是微批，不是真流。
   微批让 Spark 在容错、SQL、生态上遥遥领先，但延迟下限是百毫秒级。选型时要清楚业务延迟底线。

2. 📌Watermark = 引擎的宽容度。
   只有在 Append 模式下，Watermark 才会触发窗口关闭和 State 清理。有状态聚合的 State 无限增长，99% 是因为没有正确配置 Watermark + Append 模式。

3. 📌Checkpoint 是流作业的生命线。
   没有 Checkpoint 的流作业重启后会全量重放；多个 Query 必须有各自独立的 Checkpoint 路径；修改 Query Schema 后必须清空旧 Checkpoint。

---