9.3 事件时间、Watermark 与端到端语义：流处理怎样面对迟到与重放

批处理报告每天生成，但运维团队希望在设备温度连续十分钟异常时立即告警。网络中断后，昨晚的传感器事件今天才到；作业重启又重放了一批消息。若按到达时间统计，历史会漂移；若只求“每条恰好处理一次”，下游告警接口仍可能收到重复请求。

流处理的难点不是循环读取消息，而是对无界、乱序、可重放数据维持明确的时间和状态语义。

本课目标

区分事件时间、摄取时间和处理时间；
用窗口与 watermark 管理乱序和状态；
解释 at-most-once、at-least-once 与端到端 exactly-once 的边界；
设计可重放、幂等和可观测的流式管道。

1. 无界数据没有“全部到齐”

批表有一个可枚举快照，流则持续产生。系统必须决定：

何时触发计算；
一个窗口何时可输出初步结果；
迟到数据到来时更新、撤回还是丢弃；
状态保留多久；
失败后从哪里恢复。

流处理常以 micro-batch 或逐记录方式执行，但两者都要回答这些语义问题。

2. 三个时钟

事件时间：事件在来源系统中发生的时间；
摄取时间：平台首次接收到事件的时间；
处理时间：某个算子实际处理事件的时间。

按处理时间做十分钟窗口容易实现，但重放历史时结果会改变。业务统计通常使用事件时间，同时保留摄取时间以测量延迟。来源时钟漂移、时区和设备重启也必须进入质量监控。

3. 窗口定义是业务规则

tumbling window：不重叠固定窗口；
sliding window：按滑动步长重叠；
session window：按活动间隔动态闭合。

“过去十分钟”要说明边界、时区、触发频率和结果更新方式。一个事件可进入多个 sliding windows，状态与计算成本随重叠程度增加。

下面用 PySpark 4.0.1、受支持的 Java 17/21 和本地 JSON 文件运行一个微批查询。availableNow 处理当前可用输入后结束；临时 checkpoint 仅用于演示，生产恢复必须保留持久路径。

python

import json
from pathlib import Path
from tempfile import TemporaryDirectory
from pyspark.sql import SparkSession, functions as F

spark = (SparkSession.builder.master("local[2]").appName("atlas-stream")
         .config("spark.sql.shuffle.partitions", "2")
         .config("spark.sql.session.timeZone", "UTC").getOrCreate())
with TemporaryDirectory(prefix="atlas-stream-") as directory:
    source = Path(directory) / "input"
    source.mkdir()
    rows = [
        {"event_time": "2026-01-01T00:01:00Z", "equipment_id": "e1", "temperature": 80.0},
        {"event_time": "2026-01-01T00:05:00Z", "equipment_id": "e1", "temperature": 100.0},
    ]
    (source / "part.json").write_text("\n".join(json.dumps(row) for row in rows))
    events = spark.readStream.schema(
        "event_time timestamp, equipment_id string, temperature double"
    ).json(str(source))
    temperatures = (events.withWatermark("event_time", "30 minutes")
        .groupBy(F.window("event_time", "10 minutes"), "equipment_id")
        .agg(F.avg("temperature").alias("avg_temperature")))
    seen = []
    def inspect_batch(batch, batch_id):
        seen.extend(row.asDict() for row in batch.collect())
    query = (temperatures.writeStream.outputMode("update")
        .foreachBatch(inspect_batch)
        .option("checkpointLocation", str(Path(directory) / "checkpoint"))
        .trigger(availableNow=True).start())
    query.awaitTermination()
    assert any(row["avg_temperature"] == 90.0 for row in seen)
spark.stop()

例子计算 UTC 半开窗口 [00:00, 00:10) 内的样本均温 90，不证明温度“连续十分钟异常”。后者还需要采样覆盖、阈值持续时间与缺测规则。foreachBatch 在这里把小结果收回进程检查；不能据此声称下游持久化或故障恢复已经完成。

4. Watermark 不是“当前时间减 30 分钟”

watermark 向引擎表达对事件时间进度和允许迟到程度的策略，使系统可以清理旧窗口状态。它通常与已观察到的事件时间进展有关，不应想当然地等于墙钟时间。

设置过短：更多迟到事件无法更新结果；设置过长：状态、checkpoint 和恢复成本增加。应根据真实延迟分布、业务修正窗口和资源约束选择，并监控 watermark 进度与被拒绝/遗漏的迟到事件。

watermark 也不是完整性证明。来源可能停滞、某分区可能延迟，多个输入流还需要定义全局进度策略。

在 Spark 4.0.1 的窗口聚合语义下，迟到程度小于设定阈值的数据保证不会因此被丢弃；更迟的数据则可能处理，也可能丢弃，不能断言“超过 30 分钟必丢”。例如最大已见事件时间为 12:00，30 分钟策略对应约 11:30 的事件进度，与墙钟现在是几点无关。微批推进和多输入最小水位线还会影响实际清理时机。见 Spark 水位线保证。

5. 去重需要稳定事件身份

消息系统重投、producer 重试和回放都会产生重复。应优先使用来源生成的稳定 event_id，并定义身份作用域和保留期限。

仅按整行值去重会误删内容相同但真实发生两次的事件；仅在无限历史上保存所有 ID 又会让状态无界。结合事件 ID、事件时间和 watermark 管理去重状态，并记录超出保留窗口的重复怎样处理。

业务更新还要区分：重复、修订、撤销和晚到新事件。它们不是同一种操作。

例如 Spark 的 dropDuplicatesWithinWatermark(["event_id"]) 需要先设置水位线，并依赖重复事件时间差落在保留范围内；它不能充当永久 ID 台账。相同 ID 携带不同业务内容应按冲突或修订处理，不应直接静默丢掉。

6. 处理次数不是端到端结果语义

常见术语：

at-most-once：可能丢失，但不重试造成重复；
at-least-once：不丢为目标，失败重试可能重复；
exactly-once：在明确系统边界内，每个输入对状态/结果只生效一次。

端到端保证取决于整个链条：可重放 source、进度记录、确定性转换、checkpoint、sink 的幂等或事务提交。引擎内部宣称 exactly-once，不代表任意外部 HTTP API 也自动恰好执行一次。

一个实用写出模式是使用批次/事件幂等键：

text

begin transaction
  if output_id not committed:
      write result rows
      record output_id
commit

结果与提交标记必须原子写入同一事务边界；先写结果再单独记标记仍有崩溃窗口。

检查标记与写入还需防并发竞态，通常由唯一约束和事务隔离配合完成。下面用 Python 标准库 SQLite 演示：提交前失败会同时回滚结果与标记，提交后重试不会再追加一份；同键不同内容报错。

python

import sqlite3

connection = sqlite3.connect(":memory:")
connection.executescript("""
    CREATE TABLE committed (output_id TEXT PRIMARY KEY, value INTEGER NOT NULL);
    CREATE TABLE results (output_id TEXT PRIMARY KEY, value INTEGER NOT NULL);
""")
def publish(output_id, value, fail=False):
    with connection:
        inserted = connection.execute(
            "INSERT OR IGNORE INTO committed VALUES (?, ?)", (output_id, value)
        ).rowcount
        if not inserted:
            old = connection.execute("SELECT value FROM committed WHERE output_id=?",
                                     (output_id,)).fetchone()[0]
            if old != value:
                raise ValueError("idempotency key reused with different payload")
            return
        connection.execute("INSERT INTO results VALUES (?, ?)", (output_id, value))
        if fail:
            raise RuntimeError("simulated failure before commit")

try:
    publish("query-v1:batch-0", 90, fail=True)
except RuntimeError:
    pass
assert connection.execute("SELECT count(*) FROM results").fetchone()[0] == 0
publish("query-v1:batch-0", 90)
publish("query-v1:batch-0", 90)
print(connection.execute("SELECT * FROM results").fetchall())
# [('query-v1:batch-0', 90)]

该例只验证单数据库事务内的写入。外部 HTTP 调用不在 SQLite 事务里，需接收端幂等或事务 outbox 等独立协议。批次 ID 还要带查询/输出世代作用域；删除 checkpoint 后从 0 重新编号，不能与旧批次错误碰撞。更新窗口的结果键还需包含设备与窗口边界，并防止旧版本重试覆盖新版本。

7. Checkpoint 记录执行进度

checkpoint 常包含 source offsets、状态和查询元数据，使作业失败后继续。它不是长期业务数据备份，也不保证能跨任意代码和 schema 变更恢复。

部署变更前要确认：

新代码是否兼容旧状态 schema；
checkpoint 路径是否唯一且持久；
source 数据保留期能否覆盖最长故障；
sink 可否安全接受重放；
无法原地升级时怎样从原始日志重建状态。

保留可重放的原始事件日志，是修复逻辑错误和回填结果的关键能力。

8. Backpressure 与运行健康

输入速率长期高于处理速率时，lag 会增长。监控至少包括：

source 最新 offset 与已处理 offset；
每批输入行数、处理时长和调度间隔；
watermark/event-time lag；
state rows、state bytes 和 checkpoint 时长；
迟到、重复、解析失败与死信数量；
sink 延迟、错误和幂等冲突。

扩容前先判断瓶颈是 source 限流、shuffle、热点键、状态膨胀还是 sink 吞吐。无限增加消费并行度可能只把压力推给下游。

9. 迟到结果怎样对外发布

可选择：

追加：窗口足够确定后只写一次；延迟更高；
更新/upsert：迟到数据到来时更新同一结果键；
撤回/变更日志：发出旧值撤回与新值；
初值加终值：低延迟初步结果，随后封账并标注版本。

下游必须理解所选语义。把会更新的流写进只支持 append 的表，会产生多个互相冲突的“最终值”。

10. 测试与演练

构造包含以下情况的确定性事件序列：

text

乱序但在 watermark 内
超过允许迟到范围
相同 event_id 重投
同内容不同 event_id
作业在写出前后崩溃
source 分区暂时停滞
schema 增列与不兼容改型
从旧 offset 全量重放

验证范围包括单次输出、重启后状态、重复副作用和最终收敛结果。

常见误区

流处理就是更快的批处理：无界输入需要进度、状态和修正语义。
watermark 前的数据一定完整：它是迟到与状态的工程策略，不是事实证明。
引擎 exactly-once 等于端到端 exactly-once：sink 与外部副作用决定最终边界。
checkpoint 可以替代原始日志：逻辑错误和不兼容升级仍需要重放源。

练习

为十分钟温度告警定义事件时间、窗口、watermark 和更新语义。
生成乱序与重复事件，验证窗口结果最终是否收敛。
设计一个对作业重试安全的数据库 upsert 键。
为状态持续增长的作业画出 lag、watermark 和 state size 排障路径。

小结

可靠流处理要把时间、状态和失败当成正常输入。watermark 决定迟到与资源的权衡，checkpoint 支持恢复，而端到端交付保证必须跨过 source、引擎和 sink 的完整边界。

下一章进入数据伦理：系统能够收集、推断和自动决策，并不等于它应该这样做。

第1章逻辑

第2章集合论

第3章证明方法

第4章递归、归纳与递推

第5章关系与函数

第6章组合数学

第7章图论

第8章概率与统计

第9章数论

第10章线性代数

第11章微积分与优化

第12章信息论与数值计算