62.1 流处理模型与时间语义
一句话
流处理不是「批处理变快」，而是在数据不完整时做决策：时间语义决定一条事件属于哪个窗口，水位线决定什么时候敢宣布结果完整——两者一起决定结果的正确性与延迟。
六个误区
| 误区 | 现实 |
|---|---|
| 流处理 = 低延迟批处理 | 批处理有界、可等齐；流处理无界、永远等不齐，必须用时间语义定义「什么时候算完」 |
| 实时 = 正确 | 实时只说明「快」；用处理时间切窗口，结果会随到达顺序变化，重跑对不上 |
| 事件时间 = 处理时间 | 事件时间是业务发生时刻，处理时间是机器看到它的时刻；两者差值就是乱序/延迟 |
| 有 Kafka 就是流处理 | Kafka 是传输层；没有状态、窗口、水位与容错，消费者只是「逐条处理」 |
| exactly-once 是默认能力 | 端到端 exactly-once 要状态、位点、输出提交三方对齐（见 62.4 Exactly-once 语义） |
| 窗口结果只追加 | 迟到数据会修正历史窗口；只追加会让同一个窗口出现多条互相矛盾的记录 |
三种计算模型
| 模型 | 数据边界 | 延迟 | 状态 | 正确性来源 | 典型 |
|---|---|---|---|---|---|
| 请求-响应 | 单次请求 | 毫秒 | 无/外部 | 事务与校验 | API、交易 |
| 批处理 | 有界 | 分钟–小时 | 中间落盘 | 输入完整 + 重跑 | 离线数仓（61.3 批处理与调度） |
| 流处理 | 无界 | 毫秒–秒 | 有状态 | 事件时间 + 水位 + 快照 | 实时风控、实时数仓 |
🔑 三者的分界不是「数据量大小」，而是数据有没有界、结果什么时候必须给出。批处理可以等「输入齐了」再算；流处理没有「齐」的时刻，只能定义「等到什么时候算齐」。
流与表：同一份数据的两个视角
| 视角 | 是什么 | 例子 |
|---|---|---|
| 流 | 按时间追加的事件序列 | 订单创建、支付、退款事件 |
| 表 | 流的最新状态（changelog） | orders 表：每个订单一行，保留最新状态 |
| 流 → 表 | 按主键 upsert | order_id 相同的事件覆盖旧状态 |
| 表 → 流 | 表的变更日志 | CDC（46.4 CDC (变更数据捕获)） |
流处理里的「结果表」通常是不断更新的 changelog，不是一次性写完的文件。这也是为什么实时指标要支持 upsert/retract，而不能只 append。
三种时间语义
| 时间 | 定义 | 谁产生 | 优点 | 坑 |
|---|---|---|---|---|
| 事件时间 | 业务事件实际发生的时间 | 业务系统/客户端 | 可重放、结果稳定、能处理乱序 | 依赖业务时钟；可能缺失/漂移 |
| 处理时间 | 机器处理这条记录的时间 | 流处理器 | 实现简单、延迟低 | 结果依赖到达顺序，重跑不一致 |
| 摄入时间 | 数据进入消息队列的时间 | Kafka broker | 介于两者之间；事件时间不可信时可用 | 上游积压时，摄入时间≠业务时间 |
🔑 选时间语义的判据：这个结果需不需要重放/对账？ 需要 → 事件时间；只做「当下监控」且可接受重跑不同 → 处理时间；事件时间字段不可信但又要接近业务时间 → 摄入时间。
时间本身也会漂移：机器时钟受 NTP 调整、闰秒、虚拟化影响（见 23.1 物理时钟与漂移）。事件时间来自业务系统，处理时间来自流处理器，两者不能混用。
水位线：把「等多久」变成一个数
水位线（watermark）是一个单调递增的时间戳，含义是：
「事件时间 ≤ W 的数据，我认为已经到齐了。」
最常见的生成方式：
W(t) = max(已见事件时间) - delay
窗口在 窗口结束时间 <= W 时触发。delay 就是「愿意等多久」：
delay = 0：来一条就推进，乱序数据会被判迟到；delay = 5s：容忍 5 秒乱序；delay = 5min：结果更完整，但晚 5 分钟才出。
水位线不是「保证没有迟到」，只是「声明我等到这里了」。真正的迟到处理有三种：丢弃、允许更新（allowed lateness）、侧输出到迟到流。
窗口类型、触发与迟到处理的细节见 62.2 窗口与水位线；状态与 checkpoint 见 62.3 状态管理与容错。
代码：同一批事件，三种时间语义的结果
package main
import (
"fmt"
"sort"
)
// EventTime/ArriveTime 都是「秒」，从 10:00:00 起算。
type Event struct {
ID string
EventTime int
ArriveTime int
}
var events = []Event{
{"e1", 10, 12},
{"e2", 25, 28},
{"e3", 65, 70},
{"e4", 40, 85}, // 事件时间 10:00:40，迟到 45s
{"e5", 130, 132},
{"e6", 55, 140}, // 迟到 85s
{"e7", 190, 195},
{"e8", 100, 200}, // 迟到 100s
{"e9", 250, 255},
{"e10", 5, 260}, // 迟到 255s
}
func win(t int) int { return t / 60 * 60 }
func clock(sec int) string { return fmt.Sprintf("10:%02d:%02d", sec/60, sec%60) }
func main() {
trueCnt := map[int]int{}
for _, e := range events {
trueCnt[win(e.EventTime)]++
}
procCnt := map[int]int{}
for _, e := range events {
procCnt[win(e.ArriveTime)]++
}
fmt.Println("=== 同一批事件，三种时间语义的结果 ===")
fmt.Printf("%-9s %-6s %-6s %-6s %-6s\n", "window", "true", "proc", "et30", "et300")
type run struct {
cnt map[int]int
drop int
emitAt map[int]int
droppedIDs []string
}
runET := func(delay int) run {
order := append([]Event(nil), events...)
sort.Slice(order, func(i, j int) bool { return order[i].ArriveTime < order[j].ArriveTime })
cnt := map[int]int{}
closed := map[int]bool{}
emitAt := map[int]int{}
maxET, dropped := -1, 0
var droppedIDs []string
closeUntil := func(wm, now int) {
for s := 0; s < 300; s += 60 {
if !closed[s] && s+60 <= wm {
closed[s] = true
emitAt[s] = now
}
}
}
for _, e := range order {
w := win(e.EventTime)
if closed[w] {
dropped++
droppedIDs = append(droppedIDs, e.ID)
continue
}
cnt[w]++
if e.EventTime > maxET {
maxET = e.EventTime
}
closeUntil(maxET-delay, e.ArriveTime)
}
for s := 0; s < 300; s += 60 {
if !closed[s] {
closed[s] = true
emitAt[s] = 260
}
}
return run{cnt, dropped, emitAt, droppedIDs}
}
r30, r300 := runET(30), runET(300)
for s := 0; s < 300; s += 60 {
fmt.Printf("%-9s %-6d %-6d %-6d %-6d\n", clock(s), trueCnt[s], procCnt[s],
r30.cnt[s], r300.cnt[s])
}
fmt.Printf("%-9s %-6d %-6d %-6d %-6d\n", "total", len(events), len(events),
sum(r30.cnt), sum(r300.cnt))
fmt.Printf("%-9s %-6s %-6s %-6d %-6d\n", "late-drop", "-", "-", r30.drop, r300.drop)
fmt.Println("\n=== 每个窗口「最终结果」的产出时刻 ===")
fmt.Printf("%-9s %-11s %-11s %-11s\n", "window", "proc-emit", "et30-emit", "et300-emit")
for s := 0; s < 300; s += 60 {
fmt.Printf("%-9s %-11s %-11s %-11s\n", clock(s), clock(s+60), clock(r30.emitAt[s]), clock(r300.emitAt[s]))
}
fmt.Println("\n=== 被 ET+30s 丢弃的事件 ===")
for _, e := range events {
for _, id := range r30.droppedIDs {
if e.ID == id {
fmt.Printf(" %s 事件时间 %s，到达 %s，迟到 %ds\n",
e.ID, clock(e.EventTime), clock(e.ArriveTime), e.ArriveTime-e.EventTime)
}
}
}
fmt.Println("\n结论：处理时间把「到达时刻」当业务时间，窗口看起来均匀（每窗 2 条），但和真实业务分布不符；")
fmt.Println(" 事件时间等水位，结果正确但延迟；水位延迟越大，越完整，也越晚出结果。")
}
func sum(m map[int]int) int {
s := 0
for _, v := range m {
s += v
}
return s
}
本机实测（macOS，Apple M1，Go 1.27.1）：
=== 同一批事件，三种时间语义的结果 ===
window true proc et30 et300
10:00:00 5 2 3 5
10:01:00 2 2 1 2
10:02:00 1 2 1 1
10:03:00 1 2 1 1
10:04:00 1 2 1 1
total 10 10 7 10
late-drop - - 3 0
=== 每个窗口「最终结果」的产出时刻 ===
window proc-emit et30-emit et300-emit
10:00:00 10:01:00 10:02:12 10:04:20
10:01:00 10:02:00 10:03:15 10:04:20
10:02:00 10:03:00 10:04:15 10:04:20
10:03:00 10:04:00 10:04:20 10:04:20
10:04:00 10:05:00 10:04:20 10:04:20
=== 被 ET+30s 丢弃的事件 ===
e6 事件时间 10:00:55，到达 10:02:20，迟到 85s
e8 事件时间 10:01:40，到达 10:03:20，迟到 100s
e10 事件时间 10:00:05，到达 10:04:20，迟到 255s
结论：处理时间把「到达时刻」当业务时间，窗口看起来均匀（每窗 2 条），但和真实业务分布不符；
事件时间等水位，结果正确但延迟；水位延迟越大，越完整，也越晚出结果。
读数：
| 现象 | 数字 | 含义 |
|---|---|---|
| 处理时间每窗 2 条 | 2/2/2/2/2 | 到达被均匀摊开，看起来「很规整」，但与真实业务分布（5/2/1/1/1）不符 |
| ET+30s | 3/1/1/1/1，丢 3 条 | 等 30s 只补回一部分乱序；更晚的 e6/e8/e10 被丢 |
| ET+300s | 5/2/1/1/1，丢 0 条 | 结果完整，但所有窗口都等到 260s 才最终确定 |
| 首窗产出 | 处理时间 10:01:00 / ET30 10:02:12 / ET300 10:04:20 | 处理时间最快但不可重放；事件时间越等越完整、越晚 |
🔑 处理时间的结果依赖「事件怎么到」，事件时间的结果依赖「事件何时发生」。前者适合「现在系统还活着吗」这类监控，后者适合「昨天的 GMV 是多少」这类需要重放与对账的指标。
延迟与完整度的账
| 水位延迟 | 丢弃 | 首窗最终产出 | 适用 |
|---|---|---|---|
| 0（分析值） | 本例约 4/10（e4/e6/e8/e10） | 约窗口结束后 10s | 只做趋势、能接受丢数据 |
| 30s | 本例丢 3/10（e6/e8/e10） | 窗口结束后 72s | 一般业务实时指标 |
| 300s | 0 | 窗口结束后 200s | 财务、对账、必须完整 |
结果可见时间 ≈ 窗口结束 + 水位延迟 + 计算/排队时间。延迟每加一档，完整性上升、时效性下降；没有「又完整又即时」的免费选项。
⚠️ 在无界流里，
ET+300s不会「自己结束」。上表的 10:04:20 是流结束时的 flush；真实系统要靠空闲分区标记（idle source）或足够大的水位推进，否则一个不活跃的分区会让整个作业的水位停住。
结果更新：append 还是 upsert
端到端 exactly-once 的实现见 62.4 Exactly-once 语义；实时链路如何 upsert 到 OLAP 见 62.6 实时数仓。
| 模式 | 语义 | 下游要求 | 适用 |
|---|---|---|---|
| append | 每次触发都追加一条 | 下游自己按版本去重/取最新 | 日志、审计、不可变事件 |
| upsert | 同一 key 覆盖旧值 | 下游支持主键 upsert | 指标、看板、实时数仓 |
| retract | 先撤回旧值再发新值 | 下游支持撤回 | 聚合结果修正、Flink 的 retract 流 |
迟到数据在窗口关闭后到达时，如果开了 allowed lateness，窗口会再次触发；此时必须发更新（upsert/retract），而不是再 append 一条。只 append 的系统会出现「同一个窗口两条数」，查询端 SUM 直接翻倍——62.6 实时数仓 会用实测展示这个坑。
生产落地：指名 → 可抄 → 可验 → 可退
指名：流引擎用 Apache Flink（事件时间、水位、状态、checkpoint 的主场）；Kafka 生态内轻量场景用 Kafka Streams；托管方案用 Ververica / Amazon Managed Flink / Confluent Cloud。消息层用 Kafka（分区内有序，见 43.3 顺序与分区）。
可抄（Flink SQL 1.17+，未验证：本机未安装 Flink）：
CREATE TABLE orders (
order_id STRING,
region STRING,
amount DECIMAL(18,2),
ts TIMESTAMP(3),
WATERMARK FOR ts AS ts - INTERVAL '5' SECOND
) WITH (
'connector' = 'kafka',
'topic' = 'orders',
'properties.bootstrap.servers' = 'localhost:9092',
'properties.group.id' = 'gmv-job',
'scan.startup.mode' = 'group-offsets',
'format' = 'json'
);
-- 空闲分区 30s 后标记 idle，避免一个冷分区卡住全局水位
SET 'table.exec.source.idle-timeout' = '30s';
Kafka Streams 侧要自定义 TimestampExtractor，从业务字段取事件时间，而不是用 broker 的 record.timestamp：
props.put(StreamsConfig.DEFAULT_TIMESTAMP_EXTRACTOR_CLASS_CONFIG,
OrderTimeExtractor.class); // 返回 order.eventTime，缺失时抛异常而不是回退到处理时间
可验：本节 Go 程序输出 true/proc/et30/et300 四列，直接看出处理时间与事件时间的差异；Flink 侧看 Web UI 的 Watermark 指标与 currentInputWatermark，水位长期不动的第一嫌疑是空闲分区或某条事件时间极大。可退：调大 WATERMARK ... INTERVAL 或 allowed lateness；把作业切回处理时间只适用于「可接受重跑不一致」的监控；原始事件必须留在 Kafka（按保留期），否则任何时间语义都无法重放。
反模式
| 反模式 | 后果 |
|---|---|
| 用处理时间算财务报表 | 重跑/补数结果不同，审计对不上 |
| 水位线延迟设为 0 | 乱序事件全被判迟到，指标系统性偏低 |
| 水位线延迟设为「一天」 | 结果永远不出，实时链路退化成批处理 |
| 把摄入时间当事件时间 | 上游积压越久，指标时间越偏 |
| 迟到数据静默丢弃且无侧输出 | 丢了多少不知道，无法对账 |
| 窗口结果只 append 不更新 | 同一窗口多条，查询端重复计数 |
| 多输入算子水位取 max | 快分区把水位推过慢分区，慢分区数据被丢 |
| 不保留原始事件 | 发现口径错误也无法重放修正 |
自检
Q1 流处理与批处理的本质区别是什么？为什么「低延迟」不是本质区别？
▸ 答案：本质区别是数据有没有界。批处理的输入有界，可以等齐再算；流处理的输入无界，永远没有「齐」的时刻，所以必须用时间语义与水位线定义「什么时候算完」，并用状态保存中间结果。低延迟只是流处理的一个副产品；一个微批系统也可以低延迟，但仍然按有界批次处理。
Q2 事件时间、处理时间、摄入时间各适合什么场景？
▸ 答案：事件时间适合需要重放、对账、按业务时间聚合的指标（GMV、风控、用户行为）；处理时间适合只关心「现在系统状态」的监控（当前 QPS、当前队列长度），可接受重跑不一致；摄入时间适合事件时间字段不可信、但又希望接近业务时间的场景（客户端时钟不可靠时的日志分析），代价是上游积压会让摄入时间偏离业务时间。
Q3 为什么处理时间窗口的结果「不可重放」？用本节数字说明。
▸ 答案：处理时间按到达时刻分窗，结果依赖事件到达顺序。本节同一批 10 条事件，处理时间得到每窗 2 条；但按事件时间真实分布是 5/2/1/1/1。如果把到达顺序换一下（网络重试、积压），处理时间结果会变；事件时间结果不变。不可重放意味着补数、审计、A/B 对比都不可信。
Q4 水位延迟从 30s 调到 300s，结果和延迟各怎么变？
▸ 答案：本节 ET+30s 得到 7/10、丢 3 条，首窗在 10:02:12 产出；ET+300s 得到 10/10、丢 0，但所有窗口要到 10:04:20 才最终确定。延迟增加换来完整性，代价是结果可见时间变长；选择依据是业务能接受多晚看到「最终数」，以及迟到数据占比。
Q5 水位线能保证「没有迟到数据」吗？如果不能，迟到数据怎么处理？
▸ 答案：不能。水位线只是一个启发式声明，事件时间小于 W 的数据仍可能到达（网络延迟、上游补发、时钟错乱）。处理方式有三种：① 丢弃并计数/告警；② allowed lateness，在窗口关闭后一段时间内接受更新并重新触发；③ 侧输出到迟到流，离线修正或人工处理。三种都要能回答「丢了多少、为什么丢」。
Q6 多输入算子的水位线为什么要取 min，而不是 max？
▸ 答案：算子的输出水位代表「所有输入都至少到这里了」。取 max 会让快分区把水位推过慢分区，慢分区里事件时间较早的数据会被误判迟到而丢弃；取 min 才能保证没有输入被落下。代价是任何一个慢/空闲分区都会拖住全局水位，所以生产上要配 idle source 标记或超时。
Q7 下游消费实时聚合结果时，为什么必须支持 upsert/retract？
▸ 答案：窗口关闭后仍可能有迟到数据修正结果，或者上游重放导致同一窗口再次触发。如果下游只 append，同一窗口会出现多条记录，SUM 会把它们全部相加；支持 upsert（按窗口+维度主键覆盖）或 retract（先减旧值再加新值）才能保证最终结果唯一且正确。