62.6 实时数仓
一句话
实时数仓不是「把离线 SQL 改成 Flink SQL」，而是把 ODS→DWD→DWS→ADS 的分层、口径、幂等和回刷搬到持续更新的链路上：每条结果都要能被迟到数据修正，也要能被离线重算对齐。
Lambda 还是 Kappa
| Lambda | Kappa | |
|---|---|---|
| 链路 | 批处理（全量/历史）+ 流处理（增量）两条 | 只有流处理一条，历史靠重放 |
| 一致性 | 两套代码、两套口径，容易不一致 | 一套代码，口径天然一致 |
| 延迟 | 批链路 T+1，流链路分钟级 | 全链路分钟级 |
| 成本 | 两套集群/两套逻辑 | 一套逻辑，但重放成本高 |
| 适合 | 批逻辑复杂、流逻辑简单 | 流逻辑能覆盖全量、Kafka/Iceberg 可重放 |
🔑 选 Lambda 还是 Kappa，判据不是「新不新」，而是历史重放的成本：如果 Kafka 保留期 + 对象存储能支撑「从 0 重放」，Kappa 更简单；如果历史要跑复杂的批 SQL、重放一次要几小时，Lambda 更现实。很多团队实际是「Kappa 的实时链路 + 批处理做对账/回刷」的混合形态。
实时数仓的四层
| 层 | 存什么 | 更新方式 | 典型存储 | 保留 |
|---|---|---|---|---|
| ODS | 原始事件（订单、支付、退款） | append-only | Kafka | 3–7 天（按重放需求） |
| DWD | 去重后的明细、最新状态 | upsert（按业务主键） | Kafka compacted / Iceberg / Hudi | 数月–数年 |
| DWS | 窗口聚合（1min/5min/1h） | upsert（窗口+维度） | OLAP / Kafka | 数月 |
| ADS | 应用指标/宽表 | upsert（应用主键） | ClickHouse / Doris / StarRocks | 按合规 |
Kafka(ODS) ──> Flink(DWD: 去重/维表 join) ──> Flink(DWS: 窗口聚合)
│
├──> OLAP(ADS, upsert) ──> BI/API
└──> Iceberg(明细/回放) ──> 离线对账/回刷
🔑 实时数仓和离线数仓的分层是同一套逻辑，区别在「更新方式」：离线按分区覆盖，实时按主键 upsert。分层与口径的主场在 60.2 数仓分层 (ODS / DWD / DWS / ADS)，这里只讲实时链路的增量部分。
四个必须
窗口与水位见 62.2 窗口与水位线，端到端提交见 62.4 Exactly-once 语义。
① 事件时间，不是处理时间
实时指标必须按事件时间聚合，否则迟到/积压会让结果漂移（见 62.1 流处理模型与时间语义）。水位延迟决定「等多久」，allowed lateness 决定「关窗后还能不能修」。
② 去重：业务主键 + 版本
实时链路至少一次投递是常态：Kafka 重试、Flink 重启、上游重复推送都会带来重复。去重键要用业务主键（order_id），版本用 updated_at/CDC LSN/版本号；不能用作消息 ID（每次重投都变）。
③ Upsert：同一主键覆盖旧值
OLAP 表必须支持按主键 upsert；append-only 表会把「同一窗口的旧版本 + 新版本」都算进去。下面的实测会看到：append-only 的 gross 是 1120，upsert 是 920，差的 200 就是旧版本。
④ 对账：实时结果必须能被离线重算验证
实时链路再 exactly-once，也可能有口径 bug、维表过期、迟到丢失。必须有一条离线链路，从原始事件重算同一批窗口，与实时 OLAP 对比；差异为 0 才敢说「实时数仓可信」。
代码：实时聚合 + 迟到退款 + 去重 + 对账
package main
import (
"fmt"
"sort"
)
type Order struct {
ID string
User string
Amount int
Status string
EventTime int
Arrive int
Version int
}
type Window struct {
Start, End int
Region string
Gross, Refund, Net int
Version int
First, Last int
Closed int
}
var regionOf = map[string]string{"u1": "east", "u2": "east", "u3": "west", "u4": "west"}
// 到达顺序：o1 的重复投递在 100s；o2 的退款在 150s 才到（事件时间仍是 15s）。
var events = []Order{
{"o1", "u1", 100, "paid", 10, 12, 1},
{"o2", "u3", 200, "paid", 15, 16, 1},
{"o3", "u2", 50, "paid", 65, 70, 1},
{"o4", "u4", 300, "paid", 80, 82, 1},
{"o1", "u1", 100, "paid", 10, 100, 1}, // 重复投递
{"o5", "u1", 120, "paid", 130, 132, 1},
{"o2", "u3", 200, "refunded", 15, 150, 2}, // 迟到退款
{"o6", "u3", 80, "paid", 190, 195, 1},
{"o7", "u2", 70, "paid", 250, 252, 1},
}
const watermarkDelay = 30
type Pipeline struct {
orders map[string]Order
olap map[string]*Window
appendRows []Window
maxET int
dupSkipped int
}
func key(start int, region string) string { return fmt.Sprintf("%02d|%s", start, region) }
func (p *Pipeline) accept(o Order) bool {
old, ok := p.orders[o.ID]
if ok && o.Version <= old.Version {
p.dupSkipped++
return false
}
p.orders[o.ID] = o
p.recompute(o)
if o.EventTime > p.maxET {
p.maxET = o.EventTime
}
p.closeWindows(o.Arrive)
return true
}
func (p *Pipeline) recompute(o Order) {
start := o.EventTime / 60 * 60
region := regionOf[o.User]
var gross, refund int
for _, x := range p.orders {
if x.EventTime/60*60 != start || regionOf[x.User] != region {
continue
}
gross += x.Amount
if x.Status == "refunded" {
refund += x.Amount
}
}
k := key(start, region)
w, ok := p.olap[k]
if !ok {
w = &Window{Start: start, End: start + 60, Region: region, First: o.Arrive}
p.olap[k] = w
}
w.Gross, w.Refund, w.Net = gross, refund, gross-refund
w.Version++
w.Last = o.Arrive
cp := *w
p.appendRows = append(p.appendRows, cp)
}
func (p *Pipeline) closeWindows(now int) {
wm := p.maxET - watermarkDelay
for _, w := range p.olap {
if w.Closed == 0 && w.End <= wm {
w.Closed = now
}
}
}
func (p *Pipeline) flush(at int) {
for _, w := range p.olap {
if w.Closed == 0 {
w.Closed = at // 流结束：把剩余窗口按最终水位刷出
}
}
}
func (p *Pipeline) dump(title string) {
fmt.Printf("\n=== %s ===\n", title)
fmt.Printf("%-10s %-7s %-7s %-7s %-6s %-5s %-6s %-6s %-6s\n",
"window", "region", "gross", "refund", "net", "ver", "first", "last", "closed")
ws := make([]*Window, 0, len(p.olap))
for _, w := range p.olap {
ws = append(ws, w)
}
sort.Slice(ws, func(i, j int) bool {
if ws[i].Start != ws[j].Start {
return ws[i].Start < ws[j].Start
}
return ws[i].Region < ws[j].Region
})
for _, w := range ws {
fmt.Printf("[%02d,%02d) %-7s %-7d %-7d %-7d %-6d %-5d %-6d %-6d\n",
w.Start, w.End, w.Region, w.Gross, w.Refund, w.Net, w.Version, w.First, w.Last, w.Closed)
}
}
func (p *Pipeline) batch() map[string][3]int {
m := map[string][3]int{}
for _, o := range p.orders {
k := key(o.EventTime/60*60, regionOf[o.User])
v := m[k]
v[0] += o.Amount
if o.Status == "refunded" {
v[1] += o.Amount
}
v[2] = v[0] - v[1]
m[k] = v
}
return m
}
func main() {
p := &Pipeline{orders: map[string]Order{}, olap: map[string]*Window{}}
for _, o := range events[:6] { // 到 o5 为止
p.accept(o)
}
p.dump("迟到退款前（o5 刚写入，[0,60) 已按水位关闭）")
p.accept(events[6]) // o2 退款
p.dump("迟到退款后（o2 从 paid 改为 refunded，upsert 覆盖旧版本）")
for _, o := range events[7:] {
p.accept(o)
}
p.flush(252)
p.dump("最终结果（与批处理重算对比）")
batch := p.batch()
diff := 0
for k, v := range batch {
w := p.olap[k]
if w == nil || w.Gross != v[0] || w.Refund != v[1] || w.Net != v[2] {
diff++
}
}
gross, net := 0, 0
for _, w := range p.olap {
gross += w.Gross
net += w.Net
}
fmt.Printf("\n批处理重算：%d 个窗口，与实时 OLAP 差异 %d 个；实时总 gross=%d net=%d\n", len(batch), diff, gross, net)
ag, an := 0, 0
for _, w := range p.appendRows {
ag += w.Gross
an += w.Net
}
fmt.Printf("append-only 反例：写入 %d 行，按行求和 gross=%d net=%d（重复计算旧版本）\n",
len(p.appendRows), ag, an)
fmt.Printf("去重：跳过 %d 条重复投递；若不跳过，gross 还会多 100\n", p.dupSkipped)
fmt.Println("迟到修正：o2 的事件时间是 15s，但直到 150s 才到；[0,60) west 的版本从 v1 升到 v2。")
}
本机实测（macOS，Apple M1，Go 1.27.1）：
=== 迟到退款前（o5 刚写入，[0,60) 已按水位关闭） ===
window region gross refund net ver first last closed
[00,60) east 100 0 100 1 12 12 132
[00,60) west 200 0 200 1 16 16 132
[60,120) east 50 0 50 1 70 70 0
[60,120) west 300 0 300 1 82 82 0
[120,180) east 120 0 120 1 132 132 0
=== 迟到退款后（o2 从 paid 改为 refunded，upsert 覆盖旧版本） ===
window region gross refund net ver first last closed
[00,60) east 100 0 100 1 12 12 132
[00,60) west 200 200 0 2 16 150 132
[60,120) east 50 0 50 1 70 70 0
[60,120) west 300 0 300 1 82 82 0
[120,180) east 120 0 120 1 132 132 0
=== 最终结果（与批处理重算对比） ===
window region gross refund net ver first last closed
[00,60) east 100 0 100 1 12 12 132
[00,60) west 200 200 0 2 16 150 132
[60,120) east 50 0 50 1 70 70 195
[60,120) west 300 0 300 1 82 82 195
[120,180) east 120 0 120 1 132 132 252
[180,240) west 80 0 80 1 195 195 252
[240,300) east 70 0 70 1 252 252 252
批处理重算：7 个窗口，与实时 OLAP 差异 0 个；实时总 gross=920 net=720
append-only 反例：写入 8 行，按行求和 gross=1120 net=920（重复计算旧版本）
去重：跳过 1 条重复投递；若不跳过，gross 还会多 100
迟到修正：o2 的事件时间是 15s，但直到 150s 才到；[0,60) west 的版本从 v1 升到 v2。
读数：
| 现象 | 数字 | 说明 |
|---|---|---|
退款前 [0,60) west | gross 200 / net 200 | 只有 o2 的支付，尚未退款 |
| 退款后同窗口 | gross 200 / refund 200 / net 0，v2 | 迟到 135s 的退款仍被接受（allowed lateness） |
| 最终对账 | 7 个窗口，差异 0 | 实时 OLAP 与批处理重算一致 |
| 总量 | gross 920 / net 720 | 只有 o2 一笔退款 200；o5 是 paid，不是退款 |
| append-only | 8 行，gross 1120 / net 920 | 旧版本被重复相加，gross 多 200 |
| 去重 | 跳过 1 条重复投递 | 若不跳过，gross 再多 100 |
⚠️ 上表的净额 720 只扣了 o2 的 200 退款；
[120,180)的 o5 是 paid，没有退款。如果业务把「退款」定义成独立负向流水，而不是改订单状态，实时链路要按负向金额累加，不能只改状态——这是实时数仓最常见的口径分歧之一。
迟到数据：关窗后还能不能修
本例 [0,60) 在 132s 关闭（水位 100s），o2 的退款在 150s 才到，仍然被接受并 upsert 成 v2。原因是代码没有硬性拒绝「关窗后的迟到」，而是保留了窗口状态；生产上对应 allowed lateness：窗口关闭后状态再保留一段时间（比如 5 分钟），期间到达的数据继续更新结果。
超过 allowed lateness 的三种处理：
| 策略 | 做法 | 适用 |
|---|---|---|
| 丢弃 + 指标 | 记录 late_dropped 计数与样本 | 可接受少量丢失 |
| 侧输出 | 写到迟到流，离线修正 | 必须完整、可对账 |
| 重放 | 从 Kafka/Iceberg 重算受影响窗口 | 大范围错误、回刷 |
🔑 实时数仓的「实时」不是「一锤定音」，而是「先给一个可用的近似值，再持续修正到最终值」。下游必须能接受同一窗口的多次更新（upsert/retract），否则只能等最终值，失去实时意义。
Serving：OLAP 表怎么接住 upsert
| 引擎 | 表模型 | 去重/更新 | 查询注意 |
|---|---|---|---|
| ClickHouse | ReplacingMergeTree(version) | 后台合并按 version 保留最新；查询 FINAL 或 argMax | FINAL 有性能代价；用 argMax 更可控 |
| Apache Doris | UNIQUE KEY（Merge-on-Write） | 按主键覆盖，导入即合并 | 主键要包含窗口+维度，否则覆盖错行 |
| StarRocks | PRIMARY KEY | 按主键 upsert | 主键顺序影响前缀索引 |
ClickHouse 示例（未验证：本机未安装 ClickHouse）：
CREATE TABLE gmv_1m (
window_start DateTime,
region String,
gross Decimal(18,2),
refund Decimal(18,2),
net Decimal(18,2),
version UInt64
) ENGINE = ReplacingMergeTree(version)
ORDER BY (window_start, region);
-- 查询：同一主键取最新版本
SELECT window_start, region,
argMax(gross, version) AS gross,
argMax(refund, version) AS refund,
argMax(net, version) AS net
FROM gmv_1m
GROUP BY window_start, region;
⚠️ upsert 的主键必须包含「窗口 + 所有查询维度 + 租户」。少一个维度，不同维度的结果会互相覆盖；多一个高基数列（比如
order_id），表会膨胀成明细表，失去聚合表的意义。
对账与回刷：实时链路的刹车
| 对账 | 比什么 | 频率 | 差异处理 |
|---|---|---|---|
| 实时 vs 批处理 | 同窗口 gross/refund/net | 每小时/每天 | 差异 > 0 先查口径，再查迟到/重复 |
| 实时 vs 源系统 | 订单数、支付金额 | 每天 | 漏采、重复、状态未同步 |
| DWD vs ODS | 去重后行数、主键唯一 | 每小时 | 去重键/版本错 |
回刷（backfill）的步骤：
- 确定受影响窗口（按事件时间，不是处理时间）；
- 从 Kafka（保留期内）或 Iceberg（长期）重放；
- 用同一套 Flink SQL/代码，写同一张 OLAP 表（upsert 幂等）；
- 回刷前后各跑一次对账，差异为 0 才算完成。
🔑 回刷能力是实时数仓的「后悔药」：没有原始事件留存、没有可重放的链路，口径错了只能全量重做。Kafka 保留期与 Iceberg 明细表，至少要有一个能覆盖业务的最大回溯窗口。
生产落地：指名 → 可抄 → 可验 → 可退
指名：Flink（实时计算）+ Kafka（ODS/明细）+ ClickHouse / Apache Doris / StarRocks（ADS serving）+ Iceberg（明细/回放/对账）；调度与回刷用 Airflow/DolphinScheduler。
可抄（Doris 3.x，未验证：本机未安装 Doris）：
CREATE TABLE gmv_1m (
window_start DATETIME,
region VARCHAR(32),
gross DECIMAL(18,2),
refund DECIMAL(18,2),
net DECIMAL(18,2),
version BIGINT
) UNIQUE KEY(window_start, region)
DISTRIBUTED BY HASH(region) BUCKETS 8
PROPERTIES ("replication_num" = "3");
可验：本节程序输出「实时 OLAP vs 批处理差异 0」；生产上跑对账 SQL：
-- 同一窗口，实时与离线 gross 的差异（应为 0）
SELECT r.window_start, r.region, r.gross AS rt_gross, b.gross AS batch_gross,
r.gross - b.gross AS diff
FROM gmv_1m r
JOIN gmv_1m_batch b USING (window_start, region)
WHERE r.gross != b.gross;
可退：OLAP 表按窗口分区，可整分区重刷；Flink 作业用 savepoint 回滚；关闭实时链路时，BI 切到离线表并标注「数据延迟」；保留 Iceberg 明细，任何口径错误都能重算。
反模式
| 反模式 | 后果 |
|---|---|
| ADS 用 append-only 表 | 同一窗口旧版本 + 新版本都算，gross 虚高（本节 1120 vs 920） |
| 不去重，靠「应该不会重复」 | Kafka 重试/重启就重复，指标偏高 |
| 用消息 ID 去重 | 重投时 ID 变了，去重失效 |
| 维表 join 不带时间版本 | 退款/改地址后历史窗口被「现在」的维度改写 |
| 迟到数据直接丢且不计数 | 丢了多少不知道，对账对不上 |
| 实时与离线两套口径 | 同一个 GMV 两个数，争议时无法证明谁对 |
| OLAP 主键漏维度/租户 | 不同维度互相覆盖，数据串行 |
| 没有原始事件/回放能力 | 口径错了只能全量重做 |
自检
Q1 Lambda 与 Kappa 怎么选？判据是什么？
▸ 答案：判据是「历史重放的成本」与「批逻辑复杂度」。如果 Kafka/Iceberg 能在可接受时间内从 0 重放，且流逻辑能覆盖全量，选 Kappa（一套代码、口径一致）；如果历史要跑复杂批 SQL、重放一次几小时，选 Lambda（实时链路 + 批处理对账/回刷）。很多生产系统是混合：实时 Kappa + 离线批对账。
Q2 实时数仓的四层各存什么、怎么更新？
▸ 答案：ODS 存原始事件，append-only（Kafka）；DWD 存去重后的明细/最新状态，按业务主键 upsert（Kafka compacted/Iceberg/Hudi）；DWS 存窗口聚合，按「窗口+维度」upsert；ADS 存应用指标/宽表，按应用主键 upsert 到 OLAP。与离线数仓的分层相同，区别是「分区覆盖」变成「主键 upsert」。
Q3 本节 append-only 的 gross 为什么是 1120，而 upsert 是 920？
▸ 答案：[0,60) west 先写入 gross 200（v1），退款后又写入 gross 200 / refund 200（v2）。append-only 把两行都相加，gross 变 400；upsert 按 (window, region) 覆盖，只保留 v2 的 200。总量差 200。append-only 的 8 行里有一行是旧版本，查询端不按版本去重就会重复计数。
Q4 实时链路的去重键和版本应该怎么选？
▸ 答案：去重键用业务主键（order_id），版本用业务可比较的字段（updated_at、CDC LSN、版本号）。不能用消息 ID（每次重投都变），也不能用处理时间（同一事件重放时间不同）。同一主键只保留版本最大的记录；版本相同视为重复，跳过。
Q5 迟到数据在窗口关闭后到达，有哪几种处理方式？
▸ 答案：① allowed lateness：窗口状态保留一段时间，迟到数据继续更新并重新触发；② 侧输出：超过 lateness 的数据写到迟到流，离线修正；③ 丢弃并计数；④ 从 Kafka/Iceberg 重放受影响窗口。选择依据是「能不能接受少量丢失」与「是否必须对账到 0 差异」。本例 o2 在关窗后 18s 到达，仍在 lateness 内，所以 upsert 成 v2。
Q6 实时数仓的维表 join 为什么要带时间版本？
▸ 答案：维表属性会变（用户等级、商品类目、门店归属）。如果实时 join 只用「当前维表」，历史窗口会被新属性改写，同一个订单在不同时间跑出的结果不同。要用 temporal join（按事件时间找当时有效的维度版本），或把维度属性快照进事实表，保证结果可重放。
Q7 实时结果怎么对账、怎么回刷？
▸ 答案：对账用离线链路从原始事件重算同一批窗口，与实时 OLAP 比 gross/refund/net，差异应为 0；差异先查口径，再查迟到/重复/维表。回刷按事件时间确定受影响窗口，从 Kafka（保留期内）或 Iceberg 重放，用同一套逻辑 upsert 回 OLAP，回刷前后各跑一次对账。没有原始事件留存就没有回刷能力。