bg16a.短视频-行为链
看视频，我只看...
在本文开始之前，还是先讲讲短视频是怎么赚钱的：
| 路径 | 收入怎么产生 | 数据形状 | 本系列在哪讲 |
|---|---|---|---|
| 信息流广告 | 用户上下滑动时，每隔若干个视频插一条广告，广告主按曝光/点击付费 | 曝光流、点击流、转化流 | 17 篇（本篇只讲它与时长的关系） |
| 直播打赏 / 电商带货 | 用户下单或送礼，平台抽成 | 订单 + 状态机 + 退款 | 与 15 篇同构 |
| 会员 / 增值 | 订阅付费 | 订阅周期快照 | 15 篇 §3.3 的周期快照方法可直接套 |
广告是短视频的主收入路径，所以本篇的技术选择都被它牵着走。
先给两个必须先定义的词。
广告库存（ad inventory）：产品形态允许插入广告的位置总数。信息流短视频里，一个"位置"就是"用户滑动过程中被拿来放广告的那一刷"。库存是当天产生、当天消耗的——今天没卖掉的曝光机会，明天不能再卖。
eCPM（effective Cost Per Mille）：每一千次广告曝光带来的收入。它是把"曝光次数"换算成"钱"的乘数。它怎么被竞价机制决定，是 17 篇的内容；本篇只把它当一个已知系数用。
于是收入可以逐层代入（这是定义的展开，不是类比）：
广告收入 = 广告曝光数 × eCPM ÷ 1000
广告曝光数 = DAU × 人均刷过的视频数 ÷ 广告插入间隔
人均刷过的视频数 ≈ 人均在线时长 ÷ 人均单个视频的观看时长
代入后：
广告收入 ∝ DAU × 人均时长
结论：DAU 和人均时长是收入的直接乘数因子。 这两个数字乘起来就是"总时长"，总时长直接决定当天有多少广告库存可卖。
这就是为什么短视频公司的核心指标是 DAU × 人均时长，而不是电商那样的 GMV：电商的钱来自"用户买了多少"，短视频的钱来自"用户待了多久"。
指标决定数据形状。 因为"时长"是营收口径的指标，它就必须：
- 精确——不能靠采样估算（采样只能用于探索分析，不能用于营收报表）；
- 不能系统性偏低——丢一秒就是丢库存；
- 可归因到视频——运营要知道哪些内容在贡献时长；
- 实时可见——投放和运营要在当天调整策略，不能等 T+1。
这四条加起来，就把技术栈推到了本篇要讲的那个极限。
自问：既然收入 ∝ 曝光数，而曝光数 = 人均刷视频数 ÷ 插入间隔，那把插入间隔从"每 8 个视频插 1 条"改成"每 4 个视频插 1 条"，收入不就直接翻倍了吗？
短期会涨，长期会跌。 因果链是：插入密度提高 → 用户体验变差 → 人均时长下降、次日留存下降 → DAU 和人均时长这两个乘数一起缩 → 总库存反而变少。
这个对抗关系有两个数据层面的后果，本篇后面会用到：
- 必须有 A/B 实验平台来判断"这次改动到底是净赚还是净亏"（→ 批次三 20 篇）；
- 必须同时看短期指标和长期指标，而长期指标（如 7 日留存）天然是 T+1 甚至 T+7 才能算出来的——这就注定了实时链路和离线链路必须并存，不是"实时替代离线"。后文会讲这两条链路怎么对齐。
先看两条完全不同的链路
一个用户看一个视频，会同时产生两份数据，走两条完全独立的通道。混淆这两份数据是新人最常见的错误，所以先拆开。
先补一个名词，画像里没列：
CDN（Content Delivery Network，内容分发网络）：把同一份文件复制到全国/全球各地的边缘节点，用户从地理上最近的节点取数据，而不是每次都回中心机房。它解决的是"带宽成本 + 首帧延迟"。短视频的视频文件几乎必然走 CDN——否则中心机房的出口带宽会直接被打满。 CDN 的边缘节点会产生访问日志，这份日志里也有"谁在什么时候拉了哪个视频的哪一段"，所以它是一份独立的、口径不同的播放数据源。
flowchart LR
U["用户 APP"]
subgraph L1["链路一：视频内容（重带宽）"]
direction LR
CDN["CDN 边缘节点<br/>缓存视频分片"] --> OSS["对象存储<br/>视频源文件"]
end
subgraph L2["链路二：行为事件（重条数）"]
direction LR
GW["埋点上报网关"] --> K["Kafka<br/>播放心跳 topic"] --> F["Flink"] --> ICE["Iceberg / ByteHouse"]
end
U -- "① 请求视频分片<br/>（GB 级流量，条数少）" --> CDN
U -- "② 上报播放心跳<br/>（KB 级流量，条数极多）" --> GW
CDN -. "③ 边缘访问日志<br/>（另一份播放数据源）" .-> K
| 链路一（视频内容） | 链路二（行为事件） | |
|---|---|---|
| 瓶颈资源 | 带宽（GB/s 级） | 条数 / QPS（百万级/秒） |
| 单条大小 | MB 级（一个分片） | 百字节级（一条心跳） |
| 谁在意 | 成本团队、CDN 调度 | 数据平台、算法、运营、财务 |
| 数据平台的活 | 消费 CDN 日志做成本与质量分析 | 本篇的主战场 |
本篇只讲链路二。 链路一的数据（视频文件本身）是 16B 的主题。
"播放开始 / 播放结束"两个事件为什么不够
最自然的设计是：用户点开视频发一个 play_start，退出时发一个 play_end，时长 = play_end.ts - play_start.ts。
这个设计会让"时长"这个营收指标系统性偏低，而且偏得没法修。
自问：丢几个
play_end而已，补个默认值不就行了？
问题不在"丢了多少"，在于丢失是有偏的。play_end 收不到的场景，全都是这几类：
| 场景 | play_end 能发出去吗 |
这类用户的实际时长 |
|---|---|---|
| 用户正常滑到下一个视频 | 能 | 短 |
| APP 被系统杀掉（内存不足） | 不能 | 通常较长（看得久才被后台回收） |
| 用户直接上滑退出/锁屏 | 不一定（取决于系统给不给执行时间） | 中等 |
| 网络断开 | 不能（重试队列也可能随进程一起没了） | 不确定 |
| APP 崩溃 | 不能 | 不确定 |
因果链：play_end 的丢失率与"用户看得久不久""设备好不好""网络稳不稳"相关 → 丢失不是随机噪声而是系统性偏差 → 用任何统一的默认值去补，都会在不同人群、不同机型、不同地区之间引入不同方向的误差 → 而这个指标要拿去卖广告库存和做内容结算。
另外还有一个致命问题：在用户退出之前，你完全不知道他已经看了多久。 一个正在直播间待了 2 小时的用户，在 play_start/play_end 模型下，对实时大盘的贡献是 0——因为 play_end 还没来。实时时长大盘会在晚高峰系统性低报。
心跳模型：把"一段时长"拆成"一串脉冲"
心跳（heartbeat）：客户端在播放进行中，每隔固定时间（下文一律假设 5 秒）主动上报一条事件，说明"我还在播，播到第几秒了"。播放停止（暂停、切后台、退出、断网）就停止上报。
用户看了 20 秒然后 APP 被杀
时间轴: 0s 5s 10s 15s 20s (APP 被杀)
心跳: ●seq=0 ●seq=1 ●seq=2 ●seq=3 ●seq=4 ✗ 没有 play_end
play_start/play_end 模型算出的时长: 0 秒（因为 end 丢了） ❌
心跳模型算出的时长: (5-1) × 5s = 20 秒 ✅
关键性质：心跳模型下，丢失只会造成"少算最后不到一个心跳间隔"，而不是"整段时长归零"。 误差从"可能 100%"降到"最多 5 秒"，而且是单向的（只会少算，不会多算）——单向误差可以在口径里声明为"下界估计"，双向且有偏的误差不行。
心跳模型付出的三个代价，每一个都会变成具体的技术难题：
- 数据量按倍数放大。 一次播放从 2 条事件变成"时长 ÷ 5 秒"条事件。看 5 分钟就是 61 条。§2.5 会算这个量级。
- 必须做去重。 客户端网络抖动会重试上报，同一条心跳可能到达多次。所以心跳必须带
seq（会话内自增序号），服务端按(session_id, seq)去重。没有seq就没法区分"重试的同一条"和"真的又过了 5 秒"。 - 必须在服务端判定"会话结束"。 最后一条心跳到了之后，你不知道用户是"暂停了 3 秒还会继续"还是"已经走了"。所以需要一个会话超时规则（例如"最后一条心跳后 30 秒无新心跳 → 判定会话结束"）。这就是 Flink 里的会话窗口 / 定时器逻辑（07 篇讲过 timer 与 state）。
一条心跳长什么样
{
"event": "video_play_heartbeat",
"session_id": "a3f2-...-91", // 客户端生成，同一次播放的所有心跳共享；服务端不猜
"seq": 3, // 会话内自增，从 0 开始 —— 去重靠它
"uid": 10086, // 登录用户 id；未登录为 0
"did": "device-xxxx", // 设备 id；DAU 口径会用到（§3.4）
"video_id": 778899,
"video_len_ms": 60000, // 视频总长度，冗余进事件，避免下游关联维表
"progress_ms": 15000, // 当前播放到第几毫秒 —— 完播判定靠它
"is_auto_play": true, // 是滑到自动起播，还是用户主动点击 —— VV 口径会用到（§3.2）
"net_type": "wifi",
"client_ts": 1753900800000, // 客户端时间，可能不准（12 篇讲过）
"server_ts": 1753900800120 // 网关落地时间，用它做事件时间更稳
}
为什么把 video_len_ms 冗余进事件里？ 因为下游算完播率时需要"进度 ÷ 总长"。如果不冗余，每条心跳都要去关联视频维表——在百万 QPS 的流上做维表关联（lookup join）是很贵的。用空间换掉一次关联，是埋点设计里的常规做法。代价：视频被重新剪辑导致时长变化时，历史事件里记的是旧值。这通常可接受，因为分析要的正是"用户当时看的那个版本有多长"。
量级：和电商行为流比，差在哪
以下全是为了算式可算而设的假设值，不是任何公司的真实数据。
假设:
DAU = 2 亿
人均每日有效播放时长 = 90 分钟 = 5400 秒
心跳间隔 = 5 秒
单条心跳序列化后大小 = 200 字节
心跳条数/天 = 2 亿 × (5400 ÷ 5) = 2 亿 × 1080 = 2160 亿条/天
平均 QPS = 2160 亿 ÷ 86400 秒 ≈ 250 万条/秒
晚高峰(按 3 倍) ≈ 750 万条/秒
平均带宽 = 250 万 × 200 B ≈ 500 MB/s
晚高峰带宽 ≈ 1.5 GB/s
对比 15 篇的电商：交易流日千万行，行为流日千亿行。心跳流 2160 亿行，和电商行为流是同一个量级的上沿。
但真正的差别不是绝对量，是"量由什么决定"：
| 维度 | 电商行为流 | 短视频心跳流 |
|---|---|---|
| 产生机制 | 用户有动作才有事件 | 用户在线就有事件，不动也在产生 |
| 量级由什么决定 | 用户动作次数——有生理天花板（一个人一天点不了 1 万次） | 在线秒数 ÷ 心跳间隔——没有天花板 |
| 谁能改变量级 | 产品改交互流程（慢、需要发版） | 改一个心跳间隔配置：5 秒改成 1 秒，数据量直接 ×5 |
| 单条的信息密度 | 较高（一次加购、一次搜索都有独立业务含义） | 极低（单独一条心跳几乎回答不了任何业务问题） |
| 能否采样 | 探索分析可采样 | 营收口径指标（时长）不能采样；探索分析可以 |
| 事实表粒度 | 通常就是事件粒度 | 必须先聚合到"播放会话"粒度（§3.5） |
"改一个配置就 ×5"这件事必须被当成一类风险管理起来。 算法团队为了更细的播放曲线，把心跳间隔从 5 秒调到 1 秒，是一个客户端配置下发就能完成的动作；而下游 Kafka 分区数、Flink 并行度、存储容量全都按 5 秒的量级做的容量规划。这类"上游一个参数改动导致下游容量翻倍"的耦合，必须在埋点治理流程里设审批闸口（12 篇讲的埋点管理平台就是这个闸口该待的地方）。
15 篇的口径地狱来自"一笔订单有多个状态，你选哪个状态算 GMV"。 短视频的口径地狱来源不同："看了一个视频"这件事本身没有清晰的边界。 滑过去 0.5 秒算不算看了？自动起播算不算用户看了？拖着进度条跳到结尾算不算看完了？
每一个"算不算"都是一个口径分叉，而它们乘在一起，就是运营和算法两个团队报出两个不同数字的根因。
VV（Video View，视频播放次数）：一段时间内视频被播放的次数。它是短视频最基础的规模指标，地位相当于电商的订单数。
看起来毫无歧义，实际有三个分叉：
分叉一：自动起播算不算？ 信息流里用户滑到某个视频，播放器会自动起播。用户可能 0.3 秒后就滑走了——他根本没看见内容。
- 口径 A（曝光即 VV）：起播就算一次。数字大，但会把"划过"计入。
- 口径 B（有效播放 VV）：播放时长 ≥ 阈值（常见取 3 秒、5 秒，或"≥ 视频长度的 x%"）才算一次。数字小，但更接近"真的看了"。
分叉二：中途退出又回来，算一次还是两次？
取决于客户端有没有生成新的 session_id。这不是分析师能决定的，是埋点契约决定的——所以口径文档必须写明"以客户端 session_id 为准"。
分叉三：循环播放算几次？
短视频普遍自动循环。播完再从头播，progress_ms 会回到 0。是 1 次 VV 还是 2 次？
这三个分叉的处理原则（和 15 篇 §3.1 的结论一致）：
- 不要试图统一成一个数，而是把每个口径都物化成一个独立的、名字自解释的指标：
vv_raw（起播即计）、vv_valid_3s（≥3 秒）、vv_valid_finish（完播计一次）。 - 事实表里保留足够的原子字段（
watch_duration_ms、max_progress_ms、is_auto_play、loop_count），让任何口径都能被算出来，而不是只存一个已经加工过的vv。 - 口径写进数据字典并绑定唯一指标名（治理篇 19 会讲字典怎么落地）。
播放时长：两种算法，结果能差 3 倍
这是本篇最重要的一个建模细节，因为时长直接等于钱（§1.2）。
有两种看起来都合理的算法：
- 算法 A（心跳计数法）：
时长 = (去重后的心跳条数 - 1) × 心跳间隔 - 算法 B（进度最大值法）：
时长 = MAX(progress_ms)
下面用一段真实会带来分歧的 SQL 把差异跑出来。
-- 目标：从心跳明细聚合出"播放会话"粒度的一行
-- 引擎：Flink SQL / Spark SQL 通用语法（ROW_NUMBER 窗口函数两者都支持）
WITH dedup AS (
-- 第一步：按 (session_id, seq) 去重。客户端重试会导致同一条心跳到达多次，
-- 不去重的话时长会被直接高估 —— 这是代价 2（§2.3）的落地处理。
SELECT session_id, uid, did, video_id, video_len_ms, server_ts, progress_ms, seq,
ROW_NUMBER() OVER (PARTITION BY session_id, seq ORDER BY server_ts) AS rn
FROM ods_play_heartbeat
WHERE dt = '2026-07-29'
)
SELECT
session_id,
uid,
did,
video_id,
MIN(server_ts) AS play_start_ts,
MAX(server_ts) AS last_heartbeat_ts,
-- 算法 A：时长用心跳条数推。契约是"起播发 seq=0，之后每 5 秒一条，暂停即停发"
(COUNT(*) - 1) * 5000 AS watch_duration_ms_a,
-- 算法 B：时长用进度条位置推
MAX(progress_ms) AS watch_duration_ms_b,
MAX(video_len_ms) AS video_len_ms,
-- 完播判定：必须用进度，不能用时长（原因见下方解释）
CASE WHEN MAX(progress_ms) >= MAX(video_len_ms) * 0.95 THEN 1 ELSE 0 END AS is_finish
FROM dedup
WHERE rn = 1 -- 只保留每个 (session_id, seq) 的第一条
GROUP BY session_id, uid, did, video_id;
模拟输入（ods_play_heartbeat，两个会话，视频总长都是 60000 ms = 60 秒）：
| session_id | seq | server_ts | progress_ms | 说明 |
|---|---|---|---|---|
| S1 | 0 | 10:00:00 | 0 | 起播 |
| S1 | 1 | 10:00:05 | 5000 | |
| S1 | 2 | 10:00:10 | 10000 | |
| S1 | 3 | 10:00:15 | 55000 | ← 用户把进度条拖到了 55 秒 |
| S1 | 4 | 10:00:20 | 60000 | 播到结尾 |
| S2 | 0 | 11:00:00 | 0 | 起播 |
| S2 | 1..11 | 11:00:05 ~ 11:00:55 | 5000…55000 | 正常播完第一遍 |
| S2 | 12 | 11:01:00 | 0 | ← 自动循环，进度回到 0 |
| S2 | 13..24 | 11:01:05 ~ 11:02:00 | 5000…60000 | 播完第二遍 |
运行结果：
| session_id | 心跳条数 | watch_duration_ms_a（算法 A） | watch_duration_ms_b（算法 B） | is_finish |
|---|---|---|---|---|
| S1 | 5 | 20000（20 秒） | 60000（60 秒） | 1 |
| S2 | 25 | 120000（120 秒） | 60000（60 秒） | 1 |
为什么是这个结果：S1 的用户实际只在播放器前待了 20 秒（5 条心跳跨 4 个间隔），但他拖动进度条让 progress_ms 跳到了 60000——算法 B 把"跳过的内容"算成了"看过的时长"，高估 3 倍。S2 的用户实际待了 120 秒（看了两遍），但 progress_ms 的最大值永远不会超过视频长度——算法 B 低估一半。
由此得到两条必须分开的口径规则：
- 时长必须用算法 A（心跳计数）。它衡量的是"用户在播放器前的真实秒数"，而这正是广告库存的来源。它的误差是单向的（丢心跳只会少算），可以在口径里声明为下界估计。
- 完播必须用
MAX(progress_ms) / video_len_ms，不能用时长推。因为"时长 ≥ 视频长度"既可能是看完了，也可能是循环看了两遍前半段。
同一张会话表里，两个指标用两个不同的字段计算——这不是冗余，这是因为它们回答的是两个不同的问题。
⚠️ 算法 A 的两个已知偏差，必须写进口径文档：
- 丢心跳会少算。网络差的用户时长被低估。这个偏差方向确定（只会少），但大小与网络质量相关，所以"不同地区的人均时长对比"要额外谨慎。
- 硬编码
× 5000是个陷阱。一旦心跳间隔变更（§2.5 说过这是个配置），历史数据和新数据的算法不一致。正确做法是把心跳间隔作为字段随事件上报（hb_interval_ms），SQL 里用(COUNT(*) - 1) * MAX(hb_interval_ms)。我上面写死 5000 是为了让示例简洁，生产里不要这么写。
完播率与人均时长：两个"看起来在变、其实是结构在变"的指标
完播率 = 完播次数 ÷ VV。
自问：全站完播率从 42% 掉到 35%，是内容质量下降了吗？
大概率不是。 完播率与视频长度强负相关——15 秒视频容易看完，5 分钟视频很难看完。所以：
全站完播率 = Σ(各时长档位的完播率 × 该档位的 VV 占比)
只要内容的时长分布发生变化（比如平台在推长视频，长视频的 VV 占比上升），全站完播率就会掉，而每个档位内部的完播率可能一动没动。
所以完播率必须分时长档位看（例如 0–15s / 15–60s / 60s–5min / 5min+），全站单值只能用于同比时的粗看，且必须同时看档位分布是否变了。这类"整体指标随子群体权重漂移"的现象，是分析里的常见陷阱，本系列在治理篇（19 篇）会把它作为一类规则收进指标定义规范。
人均时长 = 总时长 ÷ DAU。分子在 §3.3 定死了，分母还需要定义：
DAU（Daily Active Users，日活跃用户数）：一天内至少有过一次活跃行为的去重用户数。
这个定义里的每个词都是口径分叉：
| 词 | 分叉 |
|---|---|
| 一天 | 按哪个时区的自然日？跨零点的会话算哪天？（通常按事件时间归属自然日，但"会话跨天"要有明确规则） |
| 活跃行为 | 启动 APP 就算？还是至少播放一个视频？还是至少有效播放 3 秒？——不同口径的 DAU 能差出百分之几到百分之十几 |
| 去重用户 | 按 uid（登录用户）还是 did（设备）？未登录用户只有 did；一个人有手机+平板会被算两个 did；一台家庭平板可能是多人共用 |
DAU 有一个决定技术方案的关键性质：它不可加。
昨天 DAU (1.9 亿) + 今天 DAU (2.0 亿) ≠ 两天的去重活跃用户数
（因为大量用户两天都活跃，被算了两次）
对比一下：时长是可加的（各分片时长相加 = 总时长），VV 是可加的，DAU 不可加。
这个区别的技术后果是决定性的：
- 可加指标可以预聚合——先按小时/按地区/按视频算好局部和，最后相加。这让计算可以任意拆分、并行、增量更新。
- 不可加指标不能预聚合。你没法"先算出每个分片的去重用户数，再加起来得到全局去重用户数"。要得到精确的全局去重值，理论上必须把所有 ID 汇总到一处比对。
而"汇总 2 亿个 ID 到一处比对"在实时链路里是什么代价、有没有绕过的办法——这就是后文要解决的问题，也是 HyperLogLog 出场的原因。
事实表粒度：为什么不能停在心跳粒度
自问：心跳明细已经落到湖里了，为什么还要再建一层"播放会话"表？直接在心跳表上算不行吗？
不行，有三个理由：
理由一：单条心跳没有业务含义。 任何一个业务问题（这个视频的完播率、这个用户今天看了多久、这个作者今天涨了多少时长）都需要先把一个会话的心跳收拢在一起才能回答。如果不物化会话层，每个查询都要自己做一遍 GROUP BY session_id ——同一份聚合被重复计算几十遍。
理由二：行数差一个数量级。 用 §2.5 的假设值：
心跳粒度: 2160 亿行/天
会话粒度: 2 亿 DAU × 人均 100 次播放 = 200 亿行/天 ← 少一个数量级
理由三：会话层可以做成"宽表"，把下游最常用的关联提前做掉。 会话粒度的一行天然可以挂上：视频维度（作者、分类、时长档位）、用户维度（新老用户、地区）、以及该次播放里发生的互动（是否点赞、是否评论、是否分享）。下游 90% 的分析在这一层就够了。
于是分层是（术语与 11 篇的分层定义一致）：
ODS ods_play_heartbeat 心跳原始明细，2160 亿行/天
├─ 保留期短（例如 7~15 天），只用于排查和重算
└─ 只做最轻的清洗（格式、去重标记）
│ 按 session_id 聚合（§3.3 那段 SQL）
▼
DWD dwd_play_session 播放会话明细，200 亿行/天 ← 事实表主力
└─ 一行 = 一次播放，含时长/完播/互动/维度
│ 按各种维度组合聚合
▼
DWS dws_video_1d / dws_user_1d / dws_author_1d
轻度汇总，供报表与自助分析直接查
ODS 层保留期为什么可以短？ 因为只要 DWD 层的会话表是从 ODS 幂等地算出来的（同样的输入必然得到同样的输出），ODS 就只是"可重算的原料"。留 7~15 天足够覆盖"发现口径写错了要重算"的响应时间。代价：超过保留期后如果发现聚合逻辑有 bug，就再也无法从原始数据修复历史——所以保留期的长短本质上是**"存储成本"与"可修复窗口"之间的取舍**，需要和治理团队一起定，不是存储团队单方面决定的。
一个容易忽略的坑：会话跨天。一个会话 23:59:50 起播、00:00:30 结束，它的 5 条心跳分布在两个分区里。按
dt分区批处理时，GROUP BY session_id会把这个会话切成两半，两边都算出一个不完整的时长。 常见处理：批处理时多读一个分区的边界数据（读dt和dt-1的最后一段），按会话的起播时间归属自然日。这个补丁很朴素，但如果不做，每天零点前后都会有一批时长被算错两次——而且这个错误在总量里占比小到不容易被发现。
电商 篇的极限在存储层（行级更新、COW/MOR、写放大）。 本篇的极限在流计算层的三个地方：吞吐、分布、以及不可加指标。
吞吐预算：从 DAU 一路算到并行度
工程里经常看到"Kafka 分区设 256、Flink 并行度设 200"这种数字，但没人说它是怎么来的。下面把它推一遍。
第一步：Kafka 需要多少分区。
分区（partition）是 Kafka 的并行单位：一个分区在同一时刻只能被消费组里的一个消费者读（基础篇 06 篇讲过这个约束）。所以分区数就是消费端并行度的上限。
输入（假设值）:
晚高峰写入 = 750 万条/秒（§2.5 算出来的）
单分区安全写入上限 = 5 万条/秒 ← 这是个保守的规划值，不是 Kafka 的物理上限；
实际上限取决于消息大小、副本数、磁盘与网络，必须自己压测
分区数 = 750 万 ÷ 5 万 = 150
留出余量（故障时的重放积压、未来增长）→ 取 256
为什么取 2 的幂、以及为什么不能想加就加？ 分区数决定了
hash(key) % 分区数的结果。一旦改变分区数，同一个 key 的历史消息和新消息会落到不同分区——"同一个 key 严格有序"这个保证就在改动的那一刻断了。所以分区数要一次规划到位、留足余量，而不是等不够了再加。这是 06 篇的 key-partition 约束在容量规划上的直接后果。
第二步：Flink 的 source 并行度。
source 并行度 ≤ 分区数。超过分区数的部分是空转 subtask——它们分不到任何分区，白占 slot（还会让 watermark 逻辑变复杂，因为空转 source 不产生 watermark，需要 idle 检测）。
所以 source 并行度最多 256。
第三步：计算并行度可以比 source 更高。
每个 source subtask 要处理 = 750 万 ÷ 256 ≈ 2.9 万条/秒
问：单个 subtask（约 1 核）能处理 2.9 万条/秒吗？
→ 完全取决于每条的处理成本。
→ 纯解析 + 转发：可以。
→ 解析 + 多次 state 读写 + 维表关联：大概率不行。
假设实测单 subtask 只能做 1 万条/秒：
需要的计算并行度 = 750 万 ÷ 1 万 = 750
但 source 并行度被分区数卡在 256
→ 在 source 之后插一次 rebalance（轮询重分发），把并行度从 256 放大到 750
rebalance：Flink 的一种分区策略，把上游每个 subtask 的数据轮询均匀分发给下游所有 subtask。它和
keyBy的区别是：keyBy保证"同 key 同 subtask"（因此可以用 keyed state），rebalance不保证任何归属，只保证均匀。 代价：rebalance 是一次真正的网络 shuffle——数据要序列化、跨节点传输、反序列化。所以只在"解析很贵、必须放大并行度"时才值得，不要无脑加。
第四步：唯一可信的健康指标是 Kafka lag，不是 CPU。
消费延迟（lag）：分区的最新 offset 减去消费组已提交的 offset，即"还有多少条没被消费"。
为什么 CPU 不可信：Flink 的反压（backpressure，07/13 篇讲过）会让慢算子把压力沿链路反向传导到 source，source 于是主动降低拉取速度。结果是所有算子的 CPU 都看起来不高、水位平稳，而数据正在 Kafka 里越积越多。 只有 lag 会诚实地涨。
监控优先级（按能发现问题的能力排序）:
1. 各分区 consumer lag 及其一阶导（在涨还是在跌）
2. checkpoint 的耗时与失败次数
3. 各 subtask 的 busyTimeMsPerSecond / 处理记录数的【分布】← 注意是分布，不是平均值
4. CPU / 内存
爆款视频：一个 key 就能把整个作业拖停
场景：某个视频突然爆了，占了全站播放量的 5%。这在短视频平台不是异常，是常态——内容消费的分布本身极度长尾。
用 §4.1 的数字（全是假设值）：
晚高峰全站 = 750 万条/秒 → 5 秒窗口内 3750 万条
并行度 = 256
均摊每 subtask = 3750 万 ÷ 256 ≈ 14.6 万条/窗口
爆款视频占 5% = 5 秒窗口内 187.5 万条，全部属于同一个 video_id
如果按 keyBy(video_id) 做窗口聚合，这 187.5 万条会全部落到一个 subtask 上，那个 subtask 要处理约 190 万条——是平均值的 13 倍。
自问：把并行度从 256 加到 1024 呢？
对这个热点 key 完全无效。 原因要讲到 Flink 的 key group 这一层。
机制：keyBy → key group → subtask
Flink 官方文档（Concepts / Stateful Stream Processing，flink-docs-stable）的原文表述：
【官方口径】Key Groups 是 "the atomic unit by which Flink can redistribute Keyed State"（Flink 能重分布 keyed state 的原子单位）；"there are exactly as many Key Groups as the defined maximum parallelism"（key group 的数量恰好等于所定义的最大并行度 maxParallelism）；运行时 "each parallel instance of a keyed operator works with the keys for one or more Key Groups"（每个并行实例负责一个或多个 key group 的全部 key）。
把这三句话串起来，链路是：
一条事件 --keyBy(video_id)--> 某个 key group --(key group 到 subtask 的分配)--> 某个 subtask
▲ ▲
│ │
key group 总数 = maxParallelism 每个 subtask 负责若干个 key group
（作业创建后固定，改不了）
关键推论（这一步是从上面三句官方表述推出来的）：
- 一个 key 整体落在一个 key group 里（key group 是"原子单位"，不会被拆开）；
- 一个 key group 在任一时刻只属于一个 subtask；
- 因此 同一个 key 的处理能力上限 = 单个 subtask 的处理能力。
加并行度只能让"更多的 key 被更多的 subtask 分担"，无法让"一个 key 被多个 subtask 分担"。 这是 keyed state 语义的硬约束——因为 keyed state 的定义就是"每个 key 一份、只能被持有该 key 的那个实例访问"，如果一个 key 被两个实例同时处理，就会出现两份互不相认的 state。
⚠️ 可信度边界：我核实到的是上面引号里的三句表述。"key 具体用哪个哈希函数映射到 key group"这一步，我引用的这个官方页面没有描述（页面里没有提到任何哈希函数或取模算法）。所以本节的推论只依赖"key group 是原子单位、一个 key group 归一个 subtask"这两条，不依赖任何具体哈希实现——这也是它成立的原因。如果你需要精确到哈希实现，请查你所用版本的
KeyGroupRangeAssignment源码。
后果：为什么"一个 key 慢"会变成"整个作业不可用"
这是最容易被低估的一环。因果链：
热点 subtask 处理不过来
→ 它的输入缓冲区满
→ 反压沿链路反向传导，上游算子和 source 一起降速 （07 篇：反压机制）
→ 整个作业的吞吐被这一个 subtask 定死（木桶效应）
→ 更糟的是：checkpoint 需要 barrier 流经【所有】subtask 才能完成 （13 篇：checkpoint 对齐）
→ 热点 subtask 的 barrier 迟迟不到 → checkpoint 耗时暴涨 → 超时失败
→ checkpoint 连续失败 → 作业失去恢复点；若此时发生故障重启，
只能从很旧的 checkpoint 恢复 → 需要重放大量数据 → 积压进一步恶化
所以热点倾斜不是"慢一点"的问题，是"作业会彻底不可用"的问题。 而且它在监控上很隐蔽：全局吞吐、平均 CPU、平均处理量全都正常，只有分布是畸形的。这就是第 3 条监控要看分布的原因。
解法：两阶段聚合的流式版本
思路和离线的加盐打散一样（基础篇 05 篇讲过）：先把热 key 拆成若干份各自算局部结果，再把局部结果汇总。但流式版有一个离线版没有的必要条件，写出来才看得清。
// Flink DataStream API（Java）。目标：统计每个 video_id 每 5 秒的播放心跳数
// 场景：video_id=778899 是爆款，占全站 5% 的量
public class TwoPhaseCount {
private static final int SALT = 16; // 每个 video_id 拆成 16 个分片；16 是拍的，见下方"怎么定"
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
DataStream<HbEvent> src = env.fromSource(/* Kafka source, 256 并行度 */ null, null, "hb");
src
// ── 第一阶段：加盐局部聚合 ──────────────────────────────
.map(e -> {
// 关键行：随机盐。同一个 video_id 的事件被均匀打散到 16 个不同的复合 key 上
e.salt = ThreadLocalRandom.current().nextInt(SALT);
return e;
})
.keyBy(e -> e.videoId + "#" + e.salt) // 带盐的 key：热点被切成 16 份
// ★ 必要条件：第一阶段【必须有窗口（或攒批）】。原因见下方解释。
.window(TumblingEventTimeWindows.of(Time.seconds(5)))
.aggregate(new CountAgg()) // 输出 Partial{videoId, cnt}，每分片每窗口 1 条
// ── 第二阶段：去盐全局汇总 ──────────────────────────────
.keyBy(p -> p.videoId) // 去掉盐：同一视频的 16 条部分结果汇到一起
.window(TumblingEventTimeWindows.of(Time.seconds(5)))
.reduce((a, b) -> new Partial(a.videoId, a.cnt + b.cnt)) // 局部计数相加
.print();
env.execute();
}
}
模拟输入（一个 5 秒事件时间窗口内）：
| video_id | 该窗口内心跳条数 |
|---|---|
| 778899（爆款） | 1,875,000 |
| 其余 1,000,000 个视频 | 合计 35,625,000（平均每个约 36 条） |
| 合计 | 37,500,000 |
运行结果（并行度 256，两种写法对比）：
| 最重 subtask 的处理量 | 是平均值的多少倍 | 第二阶段单 key 收到的条数 | |
|---|---|---|---|
直接 keyBy(video_id) |
≈ 1,890,000 | 13 倍 → 反压、checkpoint 超时 | — |
| 加盐两阶段（SALT=16） | ≈ 132,000 | 0.9 倍 → 无倾斜 | 16 条 |
为什么是这个结果 —— 也是那个"必要条件"的来处：加盐让爆款的 187.5 万条被拆到 16 个复合 key 上，每份约 11.7 万条，与全局平均 14.6 万同量级，倾斜就消失了。而第二阶段之所以不会重新变成热点，关键在于第一阶段有窗口：窗口把 11.7 万条心跳压成了 1 条部分结果，所以第二阶段的 video_id=778899 这个 key 只收到 16 条。
如果第一阶段没有窗口（比如只做一次 map 就直接 keyBy(video_id)），那 187.5 万条会原封不动地流到第二阶段——加盐等于白做。 这是流式两阶段聚合与离线版最大的差别：离线的 shuffle 天然是"攒完再发"，而流式默认是"来一条发一条"，必须显式引入窗口或 mini-batch 来创造攒批的机会。
SALT 倍数怎么定：它要大到能把最热的 key 削到平均水位，但每增大一倍，第一阶段的窗口 state 数量和第二阶段的输入条数都跟着翻倍。实践上按"最热 key 占比 ÷ 目标占比"估一个下界，再压测调整。 ⚠️ 加盐是"削平"不是"消除"：16 个带盐 key 仍然要经过一次哈希才落到 subtask，它们完全可能被哈希到少于 16 个不同的 subtask（哈希碰撞）。所以加盐后的倾斜度会显著下降但不会归零，监控还得继续看分布。
Flink SQL 里有对应的自动优化：mini-batch 攒批与 local-global 两阶段聚合可以通过配置开启（相关参数名形如
table.exec.mini-batch.*与table.optimizer.agg-phase-strategy）。参数名与默认值我没有在本次核实中逐一确认，请以你所用版本的 Flink SQL 调优文档为准。 但要注意：自动 local-global 优化对可加的聚合函数（SUM/COUNT/MAX）有效，对COUNT DISTINCT的适用性完全是另一回事——这正是下一节的问题。
这套解法对什么失效
对不可加指标失效。 把上面的 COUNT(*) 换成 COUNT(DISTINCT uid)，同样的加盐写法会直接算错：
真实情况: video 778899 被 uid={A, B, C} 看过，共 6 次播放
加盐后分片1: uid={A, B} → 局部 COUNT DISTINCT = 2
加盐后分片2: uid={B, C} → 局部 COUNT DISTINCT = 2
第二阶段: 2 + 2 = 4 ❌ 真实答案是 3（B 被算了两次）
这不是精度问题，是错误。 局部去重计数不能相加——这正是 §3.4 说的"DAU 不可加"在计算层的体现。
要修正，第二阶段就不能只收到一个数字，得收到局部的集合本身（{A,B} 和 {B,C}），才能做并集去重。但那样第一阶段就没有压缩掉任何数据量——攒批的收益消失了，加盐重新变成白做。
于是问题变成：有没有一种结构，既能被"合并"（像集合并集那样正确），又足够小（像一个数字那样便宜）？
先算清楚精确去重要花多少内存
假设要在 Flink 里维护"今日 DAU"的精确去重。最直白的做法是每个并发持一个 HashSet<Long>。
你精通 Java，所以直接按 JVM 对象布局算（开启压缩指针的 64 位 HotSpot，粗算，具体值随 JVM 实现与参数变化）：
一个 HashSet<Long> 元素的实际开销:
HashMap.Node 对象 ≈ 32 B (对象头 16 + hash 4 + key 引用 4 + value 引用 4 + next 引用 4，对齐后)
Long 对象 ≈ 16 B (对象头 12 + long 8 → 对齐到 24；小值可能命中 Long 缓存，此处按不命中算)
哈希桶数组的一格 ≈ 5.3 B (引用 4 B ÷ 0.75 默认负载因子)
─────────────────────────────
合计 ≈ 53 B ← 存一个 8 字节的 uid，花掉约 53 字节
2 亿 uid × 53 B ≈ 10.6 GB
10 GB 已经不小，但真正的问题不是这个数字，是维度组合数。实时大盘不会只要一个全站 DAU，它要按维度下钻：
维度: 地区(34) × 内容分类(20) × 新老用户(2) = 1360 个组合
一个用户会命中多个组合（1 个地区 × 1 个新老 × 他看过的若干个分类，假设 3 个）
→ 所有组合的元素数合计 ≈ 2 亿 × 3 ≈ 6 亿
→ 6 亿 × 53 B ≈ 32 GB
再加上各维度的上卷组合（只按地区、只按分类、不分维度的全站）→ 还要再涨
【推断】 state 大到这个量级后，即使换成 RocksDB 状态后端（07 篇讲过它把 state 落盘、只在内存留 block cache），路径也不通顺：每条事件都要做一次"这个 uid 见过吗"的 KV 读 + 可能的写，在 750 万条/秒的输入下，一旦 state 远超 block cache，读请求大量落到磁盘，吞吐会塌。这个判断是方向性的，具体的临界点取决于硬件、key 分布与 RocksDB 调参，需要你自己压测，我不给具体阈值。
Roaring bitmap：精确但有前置条件
bitmap（位图）去重：用一个位串，第 i 位为 1 表示"整数 i 出现过"。基数 = 位串里 1 的个数。Roaring bitmap 是工程上通行的压缩位图实现：它把整数空间按高 16 位切成块，每块根据稀疏/稠密程度自动选择"存排序数组"或"存原始位图"两种编码，从而在稀疏和稠密两种分布下都不浪费空间。（14 篇 §4.3 讲过 bitmap 在人群圈选里的用法。）
用 bitmap 存 2 亿个稠密的自增 uid：2 亿 bit ÷ 8 = 25 MB。比 HashSet 的 10.6 GB 小了约 400 倍，而且精确、支持并集/交集/差集、支持回溯成员。
看起来完美。它的前置条件是：
ID 必须是（相对稠密的）整数。
uid通常是自增主键 → 满足，bitmap 可用。did（设备号）通常是字符串 UUID → 不满足。要用 bitmap，必须先建一张全局字典把 string 映射成连续 int。而这张字典本身是个不小的工程：要全局唯一、要支持高并发写入新设备、要持久化、要在实时和离线两条链路上给出完全一致的映射（否则两边的 bitmap 不可比）。- 稀疏也是问题：如果 ID 是随机的 64 位数，位图会退化——虽然 Roaring 的自适应编码能缓解，但省不出来量级。
所以：能用 uid 就用 bitmap，需要按 did 去重就得先解决字典问题。 这个字典的构建通常归治理/公共层团队（19 篇会提），不是每个业务作业各建一份。
HyperLogLog：从零讲
它是什么（Redis 官方文档口径）：
【官方口径】HyperLogLog 是一种 "probabilistic data structure that estimates the cardinality of a set"（估计集合基数的概率性数据结构）；它 "uses up to 12 KB of memory and provides a standard error rate of 0.81%"（最多用 12 KB 内存，标准误差 0.81%）；元素很少时占用 "a lot less"；可估计到 2^64 量级的基数；
PFADD/PFCOUNT是 O(1)。
核心直觉：它不存元素，只存"你见过的最罕见事件有多罕见"。
把每个元素哈希成一个 64 位二进制串。如果哈希函数足够均匀，那么一个随机串：
以 1 开头（0 个前导零）的概率 = 1/2
以 01 开头（1 个前导零）的概率 = 1/4
以 001 开头（2 个前导零）的概率 = 1/8
以 k 个前导零开头的概率 = 1 / 2^(k+1)
反过来读这张表：要碰上"有 k 个前导零"的哈希值，大约需要看 2^k 个不同的元素。所以——
只要记住"见过的最大前导零个数 k_max"，就能反推元素个数 n ≈ 2^k_max。
见过最大 3 个前导零 → 估计约 8 个不同元素
见过最大 10 个前导零 → 估计约 1024 个
见过最大 20 个前导零 → 估计约 100 万个
见过最大 30 个前导零 → 估计约 10.7 亿个
存储成本：一个小整数。 6 bit 就能表示 0~63，足以覆盖到 2^63 量级的基数。这就是 HLL 省内存的根本原因——它把"记住谁来过"换成了"记住最罕见的那一次有多罕见"。
为什么必须分桶：单个计数器方差大到没法用。
自问：既然一个 6 bit 的数就够了，为什么 Redis 要花 12 KB？
因为单个 k_max 是极值统计量，方差极大。只要有一个元素的哈希值碰巧有 25 个前导零，估计值就从 1000 直接跳到 3355 万——错三个数量级，而且只需要一个倒霉的元素。
解法是标准的降方差手段：做很多次独立实验再平均。
把 64 位哈希值切成两段:
前 14 bit → 决定这个元素进哪个桶（2^14 = 16384 个桶）
后 50 bit → 在这一桶里数前导零，更新该桶的 k_max
最后把 16384 个桶各自的估计值做【调和平均】，再乘一个偏差修正常数，得到最终估计。
调和平均（harmonic mean）：n 个数的倒数的算术平均，再取倒数。它的性质是对极大值不敏感——某个桶给出一个荒谬的巨大估计时，它的倒数接近 0，对最终结果的拉动很小。这正是 HLL 需要的：压住个别桶的离群值。
误差与内存的对应关系（可以自己验算）：HLL 原始论文（Flajolet 等，2007）给出的标准误差是 1.04 / √m，m 是桶数。
标准误差 = 1.04 / √16384 = 1.04 / 128 = 0.8125% ← 与 Redis 官方口径的 0.81% 对上
内存 = 16384 桶 × 6 bit = 98304 bit = 12288 B = 12 KB ← 与官方口径的 "up to 12 KB" 对上
【推断】 我直接核实到的 Redis 数字只有 "12 KB" 和 "0.81%" 两个。上面的 m=16384、每桶 6 bit 是我用这两个数字反解出来的（两条算式独立地都对上了，所以我认为这个反解是对的），但我没有在 Redis 文档里直接核到桶数与桶宽。如果你要引用具体实现参数，请查 Redis 源码
hyperloglog.c。
最重要的性质：可合并（mergeable）。
两个 HLL 结构的桶是一一对应的。合并 = 逐桶取 max。
为什么这样就对：并集里"第 i 桶的最大前导零"，本来就等于两个集合各自第 i 桶最大值中较大的那个。所以——
合并后的 HLL，与"把两个集合的全部元素依次插入同一个 HLL"得到的结构完全相同。合并这一步本身不引入任何额外误差。
这一条把 §3.4 的死结解开了：DAU 不可加，但 DAU 的 HLL 可以合并。 于是可以按小时、按地区、按分类各存一个 12 KB 的 HLL，需要什么维度组合就把对应的 HLL 合并起来——预聚合重新变得可能。
回到 §4.3.1 的算式：
精确 HashSet: 1360 个维度组合 → 约 32 GB
HLL: 1360 × 12 KB → 约 16 MB ← 小约 2000 倍
用 Redis 把这个性质跑一遍（Redis 在你的画像里是"使用层熟"，所以这是成本最低的验证路径）：
# 模拟输入：两个小时各自的活跃用户。故意让 uid=10088 在【两个小时都出现】
redis> PFADD dau:h10 10086 10087 10088
(integer) 1
redis> PFADD dau:h11 10088 10089
(integer) 1
redis> PFCOUNT dau:h10 # 10 点的去重用户数
(integer) 3
redis> PFCOUNT dau:h11 # 11 点的去重用户数
(integer) 2
redis> PFMERGE dau:day dau:h10 dau:h11 # 关键行：把两个小时的 HLL 合并成一个
OK
redis> PFCOUNT dau:day # 合并后的去重用户数
(integer) 4
运行结果：PFCOUNT dau:day = 4，不是 3 + 2 = 5。
为什么是 4：PFMERGE 做的是集合并集（逐桶取 max），不是数值相加。uid=10088 在两个小时里都出现过，合并后只被算一次。这就是"不可加指标变成可合并"的全部含义——你没法把两个小时的 DAU 相加，但你可以把两个小时的 HLL 合并，再数一次。
⚠️ 这个例子里
PFCOUNT恰好是精确的（3、2、4 都对）。原因是元素极少时 Redis 用的是稀疏表示（官方口径提到元素很少时内存占用 "a lot less"），此时的计数是准的。不要因此以为 HLL 总是精确的——基数上万之后返回的就是估计值，误差在 0.81% 标准误差的量级上。用几个元素做 demo 会掩盖它的概率本质。
这三个方案容易被当成"精度换内存"的简单权衡，实际差异在能力维度上，所以逐维度列出来：
| 维度 | 精确 HashSet / 明细去重 |
Roaring bitmap | HyperLogLog |
|---|---|---|---|
| 2 亿基数的内存（粗算） | ≈ 10.6 GB | ≈ 25 MB（稠密 int） | 12 KB |
| 结果精度 | 精确 | 精确 | 估计值，标准误差约 0.81%（m=16384 时） |
| 可合并（并集） | 可（但要传整个集合，不省量） | 可 | 可，且合并不引入额外误差 |
| 可求交集 / 差集 | 可 | 可 | 不可（见下） |
| 可回溯成员（列出"是哪些人"） | 可 | 可 | 不可 |
| 对 ID 类型的要求 | 无 | 必须是较稠密的整数，字符串 ID 需先建全局字典 | 无（任何可哈希的值） |
| 典型用途 | 小基数、需要明细 | 人群圈选、留存分析、漏斗 | 大盘 DAU/UV 的实时估算、高基数维度下钻 |
| 何时不该用 | 高基数实时去重 | ID 是随机字符串且无字典 | 结算、对外披露、漏斗、人群圈选 |
HLL 不能求交集这一条，实践上最容易踩。 求交只能走容斥原理：
|A ∩ B| = |A| + |B| - |A ∪ B|
↑ ↑ ↑
三个都是带 ±0.81% 误差的估计值，误差会叠加
当 A∩B 远小于 A∪B 时（这是漏斗的常态：下一步的人数远少于上一步），
三个大数相减得一个小数 → 相对误差被放大到完全不可用。
所以：漏斗分析（本质是求"完成第 1 步 ∩ 完成第 2 步 ∩ …"的人数）绝对不能用 HLL。 必须用 bitmap（对 uid）或精确明细。15 篇 §3.5 讲的漏斗就属于这一类。
HLL 不能回溯成员这一条，也有直接的业务后果：运营说"把昨天看过这个视频的人圈出来做定向推送"——HLL 存的只是寄存器，给不出名单。这类需求必须用 bitmap 或明细表。
工程上的常见分工（这是行业通行做法）：
实时大盘的 DAU/UV → HLL（在 Flink 里维护，或写入 Redis/OLAP 的 HLL 列）
★ 前端展示必须打标"预估值"
人群圈选 / 留存 / 漏斗 → Roaring bitmap（对 uid），离线构建
财务结算 / 创作者分成 → 精确明细，离线，可审计、可追溯到单条会话
实时和离线两条链路必须并存。并存就必然对不上，所以对账不是可选项。
差异从哪来
| # | 来源 | 差异方向 | 好不好查 |
|---|---|---|---|
| 1 | 迟到数据。实时按 watermark 关窗（07 篇），关窗后到达的数据被丢弃或进侧输出流；而离线 T+1 跑的时候数据已经到齐 | 实时系统性偏低 | 好查（差异集中在特定时段） |
| 2 | 口径实现不一致。两套代码（Flink SQL / Spark SQL）：UDF 行为、时区、NULL 参与聚合的方式、去重条件、边界是否闭区间，任一处不同就分叉 |
方向不定 | 最难查 |
| 3 | 去重算法不同。实时用 HLL 估计值，离线用精确值 | 随机，量级在标准误差附近 | 好查（只影响去重类指标） |
侧输出流（side output）：Flink 提供的旁路输出通道。窗口可以配置把迟到数据发到侧输出流而不是直接丢弃，这样至少能量化"到底迟到了多少"。如果连侧输出流都不配，差异 1 就变成了一个不可观测的黑洞——这是很值得提前做的一件小事。
对账任务：用"差异的分布形态"定位根因
-- 每日自动对账：昨日实时链路的最终值 vs 离线精确值，按小时拆开看
-- 关键：不能只看总量差异，要看差异【怎么分布】—— 分布形态直接指向根因
SELECT
o.stat_hour,
o.vv_offline,
r.vv_realtime,
ROUND((r.vv_realtime - o.vv_offline) * 100.0 / o.vv_offline, 3) AS vv_diff_pct,
ROUND((r.dur_realtime - o.dur_offline) * 100.0 / o.dur_offline, 3) AS dur_diff_pct
FROM dws_video_play_1h_offline o -- 离线 T+1 产出，精确，作为基准
LEFT JOIN dws_video_play_1h_rt r -- 实时链路当天产出的最终快照
ON o.dt = r.dt AND o.stat_hour = r.stat_hour
WHERE o.dt = '2026-07-29'
ORDER BY o.stat_hour;
模拟输入 / 运行结果（两种不同的故障形态，同一段 SQL 跑出来）：
形态 A：
| stat_hour | vv_offline | vv_realtime | vv_diff_pct |
|---|---|---|---|
| 00 ~ 21 | … | … | -0.02% ~ +0.03% |
| 22 | 1,820,000,000 | 1,818,500,000 | -0.08% |
| 23 | 1,650,000,000 | 1,514,700,000 | -8.20% |
形态 B：
| stat_hour | vv_offline | vv_realtime | vv_diff_pct |
|---|---|---|---|
| 00 ~ 23 | … | … | 每个小时都在 -2.4% 左右 |
为什么这两个结果指向完全不同的根因：
- 形态 A：差异集中在最后一两个小时，且方向是实时偏低。这是差异来源 1（迟到数据）的典型指纹——23 点的数据有一部分在零点后才到，实时链路已经关窗了，而离线跑的时候它们已经落库。同时也要检查 §3.5 说的跨天会话处理是否正确，它也会集中体现在 23 点和 0 点。
- 形态 B：差异在所有小时均匀分布，说明它与"数据到得早晚"无关，只与"怎么算"有关 → 指向差异来源 2（口径实现不一致）。这时该做的是逐字段比对两套 SQL：过滤条件、时区、
NULL处理、去重键、时长算法（两种算法有没有一边写错）。
均匀分布 vs 集中在尾部——这个判据能省掉大量瞎猜。 如果差异只出现在某个地区或某个 APP 版本上，则指向某个版本的客户端埋点有问题（12 篇的埋点质量监控该接上）。
处理原则
- 以离线为准，写进制度。 财务结算、创作者分成、对外披露、KPI 考核，一律用离线值。
- 实时数字必须打标"预估值"，并显示口径版本号。这一条是治理要求，不是技术要求——不打标，业务方就会拿实时数字去开会，然后第二天发现和离线不一样，信任一次就崩了。
- 对账任务化 + 阈值告警。差异率超阈值（例如 0.5%）自动告警，并按分布形态自动给出初判方向。
- 接受 Lambda 架构的代价。 两条链路 = 两套代码 = 两处都要改 = 必然会分叉（11 篇提过 Lambda 架构）。流批一体（Flink 的批流统一 SQL、Paimon/Iceberg 作为统一存储）想消掉这个代价，路径是"同一份 SQL 跑两种模式"，代价是要接受统一存储的性能与成熟度约束（13/14 篇讨论过）。这是一个仍在演进的方向，我不断言它现在就能完全替代 Lambda。
上游：我们依赖谁
| 上游 | 提供什么 | 出问题时的表现 |
|---|---|---|
| 客户端 SDK 团队 | 心跳契约的所有者：间隔、seq、session_id 生成规则、暂停/切后台是否停发 |
改了间隔 → 下游容量翻倍；改了 session 规则 → VV 口径静默漂移 |
| 上报网关 | server_ts、限流、鉴权 |
限流丢弃 → 时长系统性偏低，且偏低集中在高峰 |
| CDN | 边缘访问日志 | 可作为播放量的交叉验证源（口径不同，但趋势应一致） |
| 视频内容库 | video_id 的作者、分类、时长等维度 |
维度缺失 → 会话宽表关联不上，下钻失效 |
下游：谁在吃我们的数据，以及他们各自把什么约束加在我们头上
这一列"技术约束"才是重点——下游的需求决定了我们必须建成什么样。
| 消费方 | 要什么 | 延迟要求 | 精度要求 | 加在我们头上的约束 |
|---|---|---|---|---|
| 推荐算法（在线特征） | 用户近 N 次播放序列、视频近 1 小时的完播率 | 秒级 | 允许估算 | 最苛刻的消费方：要求实时链路端到端秒级，且要有低延迟的特征存储（→ 20 篇特征平台） |
| 广告投放 | 库存预估、流量分配、时长趋势 | 分钟级 | 中等 | 实时大盘必须稳定可用；口径必须与广告侧对齐（→ 17 篇） |
| 内容运营 | 爆款发现、内容池调控、异常内容定位 | 分钟级 | 中等 | 需要高维度自助下钻 → OLAP 引擎 + 预聚合（HLL 在这里派上用场） |
| 创作者中心 | 每个作者看自己的播放/时长/涨粉 | 分钟到小时 | 要与结算一致 | 高并发点查：数百万创作者各查自己的一行 → 14 篇讲的服务化交付形态 |
| 审核 / 风控 | 刷量识别（同一 did 异常心跳、时长分布异常、机器人特征） |
准实时 | 高 | 需要保留明细（聚合值查不出刷量），且需要 ODS 层可回查 → 影响 §3.5 的保留期决策 |
| 财务 / 创作者分成 | 按时长/播放量分成的结算数据 | T+1 即可 | 必须精确、可审计、可追溯到单条会话 | 最硬的约束：禁止 HLL 估计值、禁止采样、必须离线精确、必须可重算且重算结果稳定（幂等） |
| 成本团队 | CDN 带宽 ÷ 播放量的比值 | 小时级 | 中等 | 需要把两条链路的数据（CDN 日志 + 埋点）在同一口径下并列——这本身是个口径对齐工作 |
| A/B 实验平台 | 按实验分组的时长/留存/完播 | T+1 | 高 | 事实表必须带实验分组标识，且分组标识要在会话粒度上稳定 |
两个观察，比表格本身更重要：
- 精度要求最高的（财务）延迟要求最低，延迟要求最高的（推荐）精度要求最低。 这个负相关关系不是巧合，它就是实时/离线双链路能够长期共存的根本原因——如果有一个消费方同时要求"秒级"和"精确到分"，Lambda 架构就撑不住了。（17 篇的广告计费恰好是那个"两个都要"的场景，所以它是流处理的天花板。）
- 风控要明细、财务要可重算，这两条一起把"ODS 层能不能只留 3 天"这个看似纯成本的问题，变成了一个需要多方参与的决策。
边界与坑
- 把心跳间隔写死在 SQL 里（
× 5000）。间隔一改，历史与新数据的算法就不一致。应把间隔随事件上报。 - 用
MAX(progress_ms)算时长。拖动进度条会高估（示例里高估 3 倍），循环播放会低估（低估一半）。时长用心跳数推，完播用进度推。 - 忘了按
(session_id, seq)去重。客户端重试会让时长直接高估，且高估集中在网络差的人群。 - 全站完播率单值当质量指标看。它会随内容时长分布漂移，必须分时长档位。
- 会话跨天被切成两半，两边各算一个不完整时长。批处理要多读边界分区，按起播时间归属自然日。
- 加盐两阶段聚合在第一阶段不加窗口。等于白做——热点原封不动流到第二阶段。
- 对
COUNT DISTINCT做"局部计数相加"。这是错误不是精度损失。 - 用 HLL 的结果做结算、分成或对外披露。它是估计值，且不可审计。
- 用 HLL 做漏斗。HLL 不支持交集，容斥原理会把误差放大到不可用。
- 只看平均值不看分布。热点倾斜时全局吞吐和平均 CPU 都正常，只有 subtask 的处理量分布是畸形的；而反压会让 CPU 看起来更"健康"。
- 不配侧输出流收集迟到数据。差异 1 变成不可观测的黑洞，对账时无从下手。
- 实时数字不打标"预估值"就给业务方看。技术上没错，但信任崩一次就很难重建。
小结
本篇的一条主线：短视频的钱来自"用户待了多久" 那个恒等式），于是"时长"成了营收口径指标；营收口径要求精确、不能系统性偏低、要实时——这三条把数据模型逼成了心跳模型；心跳模型带来 2000 亿条/天的量级和极低的单条信息密度，于是事实表必须先收拢到播放会话粒度；而"看了""看完了""日活"这三个概念本身没有天然边界，于是又是一遍口径分叉。
技术上的三个极限，以及各自的破法与代价：
| 极限 | 根因 | 破法 | 代价 |
|---|---|---|---|
| 吞吐 | 750 万条/秒 | 分区数一次规划到位；source 后 rebalance 放大计算并行度 | 分区数事后改不了（破坏 key 有序）；rebalance 是一次真 shuffle |
| 分布 | 爆款视频独占 5% 流量，而一个 key 只能被一个 subtask 处理（key group 是原子单位） | 加盐两阶段聚合，第一阶段必须有窗口 | 延迟增加；state 变多；对不可加指标直接失效 |
| 不可加 | DAU 不能相加，精确去重内存 32 GB 起 | HyperLogLog：12 KB、0.81% 标准误差、合并不引入额外误差 | 是估计值；不能求交集、不能回溯成员、不能用于结算 |
一句话记住 HLL：它不记住谁来过，只记住"最罕见的那一次有多罕见"——所以它极省内存、可以合并，也因此永远给不出名单、永远算不了交集。