Lecture 21: Stream Processing and Cluster Scheduling — 流处理与集群调度

目录 · ← l20 · l22 →

第八部分：现代分布式系统与真实世界

这一部分把前面所有原语组装成真实系统，并走向课程的前沿：流处理与集群调度、分布式文件系统、分布式共享内存、图处理与分布式机器学习、安全，最后是真实数据中心灾难案例——看看这些理论在现实中如何失效。

Lecture 21: Stream Processing and Cluster Scheduling — 流处理与集群调度

讲义对应：CS 425 FA2026 Lecture 21（本笔记编号）。本章对应课程 Lecture 23「Stream Processing and Scheduling」（2026-11-10，The New Age 模块，课程表指定的阅读材料为 Storm、Spark Streaming、DRF）。主要素材：Lecture 22-B「Stream Processing」（L22.B.FA25.pdf，23 页：为什么需要流处理、Storm 的 tuple/stream/spout/bolt/topology、三种 grouping、Nimbus/Supervisor/ZooKeeper 架构、Anchoring 与失败重放、OutputCollector API、Twitter Heron 的背压优化）、Lecture 23「Scheduling」（L23.FA25.pdf，34 页：为什么需要调度、单处理器 FIFO/STF/Round-Robin、Hadoop Capacity 与 Fair Scheduler、任务长度估计、Dominant Resource Fairness 及其数值例子）、Lecture 26-B「Apache Spark」（L26.B.FA25.pdf，17 页，课程表规定所有学生必须观看该 Spark 视频，因此在考试范围内：MapReduce 为何不够用、RDD 的三条性质、lineage 容错、partitioning、GraphX GAS）。 论文素材：任务给出的 papers/drf-eecs259.pdf 经提取后确认是 Berkeley 技术报告 UCB/EECS-2012-259《Discretized Streams: A Fault-Tolerant Model for Scalable Stream Processing》（Zaharia et al., 2012），即 D-Streams / Spark Streaming 的原始论文（不是 DRF 论文；DRF 的原始文献是 Ghodsi, Zaharia, Hindman, Konwinski, Shenker, Stoica, Dominant Resource Fairness, NSDI 2011）。本章用该报告支撑 21.2.6/21.2.7 与 21.3.2/21.3.3 的流处理容错细节（并行恢复 vs 上游备份、lineage cutoff、微批、reduceByWindow、exactly-once 语义、恢复时间公式 $t_{up}=\lambda/(1-\lambda)$ 与 $t_{par}=\lambda/(N(1-\lambda))$、60M records/s、与 Storm 的吞吐对比、检查点间隔实验），DRF 部分以课程讲义口径（含 18 CPU/36 GB 数值例子）+ NSDI 2011 的四条公理为准。 教材对应：Coulouris 5th Ed. Ch. 21（Designing Distributed Systems，分布式系统设计中的云与集群管理）、Ch. 2（System Models，云与数据中心架构）；补充：Ch. 5/Ch. 12（MapReduce/GFS 背景，见 Lecture 5、Lecture 11）、Ch. 14（Time and Global States，见 Lecture 12 的 Chandy-Lamport 快照——本章 Flink barrier checkpoint 的理论基础）、Ch. 18（Replication）。 阅读材料：Zaharia et al., Discretized Streams（UCB/EECS-2012-259，本章论文素材）；Ghodsi et al., Dominant Resource Fairness, NSDI 2011；Hindman et al., Mesos, NSDI 2011；Schwarzkopf et al., Omega, EuroSys 2013；Verma et al., Borg, EuroSys 2015；Zaharia et al., RDDs, NSDI 2012；Kulkarni et al., Twitter Heron, SIGMOD 2015。

21.1 概述

本章把两个看似无关、实则同源的问题放在一起讲：（1）如何对无界（unbounded）、持续到达的数据流做低延迟计算？ 这是流处理（stream processing），其代表系统是 Storm、Spark Streaming、Flink、Kafka Streams；（2）如何在成百上千个作业共享一个数据中心的条件下，把多维资源（CPU、内存、磁盘、网络）公平且高效地分出去？ 这是集群调度（cluster scheduling），其算法核心是 DRF（Dominant Resource Fairness，主导资源公平），其代表系统是 Mesos、YARN、Borg、Kubernetes。

它们同源之处在于：两者都在回答「当计算变成一个长期运行、不断演化的分布式系统时，容错与公平如何重新定义」。Lecture 5 的 MapReduce 已经给出黄金法则——用「重新执行（re-execution）」代替「恢复状态」（任务确定性 + 输入可重放 ⇒ 重跑结果不变）。本章把它推到两个新战场：

流处理的战场：Storm 用 tuple 级确认（XOR acking）+ 超时重放，只保证 at-least-once；Spark Streaming 用离散化流（D-Streams）把流切成微批、每个批是一个不可变 RDD，靠血缘做分区级重算；Flink 用 Chandy-Lamport barrier checkpoint 做到 exactly-once。共同点：都不复制数据，而是在需要时重算数据——粒度从「整个 job」缩到「一个分区/一个算子状态/一棵 tuple tree」，恢复时间从分钟压到亚秒。
集群调度的战场：单资源公平（max-min）早有标准答案，但真实作业需求是多维的（机器学习吃 CPU、内存数据库吃内存、日志分析吃磁盘 IO）。只按 CPU 公平会把内存浪费掉，只按内存公平会把 CPU 浪费掉；DRF 的洞察是把公平重新定义在主导份额这个标量上——让所有用户「最缺那种资源的占比」尽量相等。

一句话概括本章的黄金法则（贯穿全章，21.6 再次点题）：

流处理的核心是把”容错”从”复制数据”变成”重算数据”（血缘 / barrier checkpoint），这与 MapReduce 的 re-execution 是同一种哲学，只是粒度更细、延迟更低；而集群调度的核心是”多资源公平”——DRF 用主导份额把单一资源的 max-min 公平推广到了多维资源。

本章结构：21.2 建立概念体系（批→流、窗口/水位线/状态、Storm、RDD 与血缘、Spark Streaming、Flink barrier、调度框架演进、DRF 的主导份额）；21.3 给出五份伪代码与完整正确性论证；21.4 用三个可运行程序把机制跑出来；21.5 分析性能与三方权衡；21.6–21.8 是结论、陷阱与思考题。

21.2 核心概念与分布式机制图解

21.2.1 从批处理到流处理：有界数据 vs 无界数据（Bounded vs Unbounded Data）

定义与目的：批处理（batch processing）处理的是有界数据（bounded data）——数据集在开始计算前就已经完整存在（例如 HDFS 上的一个文件），因此计算有明确的”开始”和”结束”，结果在作业结束时一次性产出。流处理（stream processing）处理的是无界数据（unbounded data）——数据是连续到达、永不结束的序列（Kafka topic、点击流、传感器遥测、日志），因此”作业永不结束”，结果必须持续地、低延迟地产出。课程讲义对”流处理挑战”的表述是：处理大量数据，延迟控制在几秒之内，同时保持高吞吐（process large amounts of data, with latencies of few seconds, with high throughput）。
直观解释（”它是什么？”）：批处理像水库放水发电——先把水蓄满（数据落盘），再一次性放下来算（一个 MapReduce 作业）；流处理像自来水厂的实时加氯——水一直在流，你必须在它流过管道的那几秒钟内完成检测和调节，而不能等它流完（永远流不完）。另一个更贴近生活的类比：批处理是月末结账，流处理是收银台的实时流水；报表可以等一个月，但信用卡盗刷必须在几秒内发现。
MapReduce（Lecture 5）在流场景下的三重局限（讲义”MapReduce?”一页的原口径是”批处理需要等待整个大数据集上的计算完成，不适合长时间运行的流式处理”）：
1. 延迟高（分钟到小时）：作业提交 → 调度 → 启动 JVM → 读 HDFS → shuffle 落盘 → reduce → 写回 HDFS，每一步都以秒到分钟计，而且为了容错中间结果必须落盘（Lecture 5）。这与「秒级出结果」差 2–3 个数量级。
2. 不适合连续到达的数据：MapReduce 的输入语义是有界的（一个已存在于 HDFS 的目录），无法「永远运行、每来一条算一次」；把流切成小文件反复提交小作业，作业启动开销会让端到端延迟退化到几十秒。
3. 不适合迭代与交互：迭代式算法（机器学习、PageRank）每轮都要读写整个数据集；交互式查询要求亚秒响应，而 MapReduce 连作业启动都不止一秒。讲义给的根因很直接：为容错而昂贵的磁盘写入导致缺乏高效的数据共享，于是迭代式与交互式应用代价极高，业界只好为每种编程模型各造轮子（Pregel、HaLoop），每个都要重新解决容错。
机制图解：两种数据处理模型的时间线对照。

   批处理 (MapReduce)                              流处理 (Stream Processing)
   ────────────────────────                        ─────────────────────────────
   数据必须有界（先落盘再算）                        数据无界、连续到达，永不"结束"

   HDFS:  ██████████████  一个 job 的输入           事件流: ▓▒░▓▒░▓▒░▓▒░▓▒░▓▒░▓▒░  →→∞
           │                                              │  (每毫秒都有新记录)
           ├─ 提交/调度(秒~分)                             v
           ├─ map 阶段(分钟)                          ┌──────────────┐
           ├─ shuffle 落盘(分钟)  ← 容错靠"落盘"       │  持续运行的   │
           ├─ reduce 阶段(分钟)                        │  算子拓扑     │
           └─ 输出(分钟)                               └──────┬───────┘
                                                             │ 秒级/亚秒级
   端到端: 分钟 ~ 小时                                 输出: ├─窗1─┤├─窗2─┤├─窗3─┤ ...
   语义:   一次作业 = 一个完整结果                      端到端: 毫秒 ~ 秒
   容错:   任务重执行（分钟级恢复）                     容错:   血缘重算/checkpoint（亚秒级恢复）

典型应用（讲义列举 + 课程思考题”下面哪个不是流处理作业？答案是全都是“）：
- 实时监控与告警：入侵检测系统（IDS）需要在数据中心的流量里发现异常模式；
- 实时搜索与趋势：Twitter 的实时搜索、社交网络热点趋势；
- 网站统计：Google Analytics 式的实时报表；
- 实时推荐 / 在线广告：Flipboard 定制化 feed、广告点击统计与竞价；
- 金融风控：TripAdvisor 的”每日营收 + 欺诈检测”（讲义思考题的选项之一）；
- IoT 遥测 / 日志分析与集群监控 / 实时个性化：Mobile Millennium 用出租车 GPS 数据估计路况、在数百节点上挖掘程序日志定位故障（都是 D-Streams 论文的真实应用）、Uber 动态定价、Netflix 用户行为分析。
关键假设与系统模型：流处理系统的目标延迟决定了它的设计取舍。D-Streams 论文明确划定范围：目标延迟 0.5–2 秒，刻意不针对亚毫秒级场景（如高频交易）。这条边界很重要：延迟目标在秒级时微批是划算的（可复用批处理成熟的确定性容错机制）；在毫秒级时就必须走逐记录 + 轻量 checkpoint 的路线（21.2.6 与 21.5 会反复回到这条边界）。

21.2.2 流处理的六个关键概念：事件时间、窗口、水位线、状态、背压、投递语义

事件时间 vs 处理时间（Event Time vs Processing Time）：
- 事件时间是事件实际发生的时刻（用户点击链接的那一刻），由产生数据的设备写入记录；处理时间是事件被系统处理的时刻（算子读到它时的本地时钟）。二者之间存在乱序（out-of-order）与迟到（late）：网络抖动、客户端离线缓存、分区切换都会让一条 10:00:01 的事件在 10:00:09 才到达。
- D-Streams/Spark Streaming 按「到达时间」切批（保证总能按时启动新批；前提是同集群节点经 NTP 同步、偏差毫秒级）。需要按事件时间分组时，论文给出两条路径：等待固定的「松弛时间（slack time）」，或应用层用增量 reduce 修正（先在 t+1 给出 [t,t+1) 的初步计数，之后把迟到记录补进去）。这与 Flink/Beam 的 watermark + allowed lateness + 触发器是同一件事的两种表述。
- 水位线（watermark）的严格定义：一条随时间单调不减的”事件时间进度声明” $W(t)$，含义是”事件时间不超过 $W(t)$ 的数据，系统认为已经全部到达“。水位线是”用延迟换确定性”的旋钮：水位线越保守（等待越久），结果的完整性越好、延迟越高。水位线不是”数据都到了”的证明，而是”我不再等了”的承诺——这是最容易误解的一点（见 21.7）。
窗口（Window）：把无界流切成有限的可计算单元。三种基本形态（配合 ASCII 图）：

   事件流（按事件时间编号）：
     e1  e2  e3  e4  e5  e6  e7  e8  e9  e10 e11 e12
      1   2   3   4   5   6   7   8   9   10  11  12

   ① 滚动窗口 (tumbling, size=4)：不重叠、不留空隙，每个事件属于恰好一个窗口
     [ e1 e2 e3 e4 ][ e5 e6 e7 e8 ][ e9 e10 e11 e12 ]
           W1              W2               W3          触发时刻: 窗口右边界到达时

   ② 滑动窗口 (sliding, size=4, slide=2)：窗口重叠，每个事件属于多个窗口（可增量计算）
     [ e1 e2 e3 e4 ]
          [ e3 e4 e5 e6 ]
               [ e5 e6 e7 e8 ]
                    [ e7 e8 e9 e10 ]
     优化: 若聚合函数既有结合律又有逆元（count/sum），用"加入新批、减去旧批"避免重算

   ③ 会话窗口 (session, gap=2)：按"活动间隔"切分，边界由数据本身决定（长度不固定）
     [ e1 e2 e3 ]   gap>2   [ e6 e7 ]        gap>2        [ e10 ]
       会话1                   会话2                        会话3

会话窗口的关键差异：前两种窗口的边界由时间决定（可预先规划），会话窗口的边界由数据决定（要等到”gap 时间内没有新事件”才能关闭窗口，天然依赖水位线或超时）。
状态（State）与状态后端：有状态算子必须记住跨记录/跨批的信息（计数、去重集合、会话表、模型参数），由此带来两个新问题：(1) 容量（状态可能超过内存，需要落盘，如 RocksDB 状态后端）；(2) 一致性（状态必须与「输入处置进度」一起被快照，否则恢复后会出现「输入重放而状态没退回去」的重复计数）。21.2.7 会看到：Flink 的 barrier 对齐本质上就是给「输入进度 + 算子状态」拍一张一致快照（Lecture 12 的 Chandy-Lamport）。
背压（Backpressure）：下游处理不过来时上游必须减速，否则缓冲区无限堆积直到内存耗尽、延迟爆炸。讲义用 Twitter Heron 给出三条实现路径：TCP 背压（滑动窗口）、Spout 背压（停止读取上游）、逐级背压（按 stage 传播）。背压是流处理系统稳定性的第一道防线，也是 21.5 讨论 Flink 对齐式 checkpoint 停顿的舞台。
投递语义（Delivery Semantics）：三种语义必须严格区分，因为它们是考试与工程中最常被混淆的一组术语。

语义	精确含义	故障时的表现	实现代价	代表
至多一次（at-most-once）	每条记录最多被处理一次，允许丢失	故障后不重放	最低（不做任何确认）	S4、纯 UDP 管道
至少一次（at-least-once）	每条记录至少被处理一次，可能重复	故障后重放 ⇒ 重复结果	中（确认 + 重放，见 tuple 级 ack）	Storm（默认）、Kafka 消费者手动提交
恰好一次（exactly-once）	每条记录对状态的影响恰好一次，端到端不重不漏	故障后回滚到一致快照再重放，重复部分被去重/覆盖	高（快照 + 事务性输出，或幂等输出）	Flink、Spark Streaming（配合幂等/事务输出）、Kafka Streams

一个必须澄清的细节：「恰好一次」通常指「状态更新的效果恰好一次」，而不是「记录恰好被投递一次」。物理上记录仍可能被重放，系统通过回滚状态 + 重放到一致点 + 幂等/事务输出让最终效果等价于恰好一次（故也称 effectively-once）；21.3.3 会给出它的三个充分条件。

21.2.3 Apache Storm 的编程模型：Topology / Spout / Bolt / Tuple / Grouping

定义与目的：Storm 是 Twitter 开源的分布式实时计算系统（讲义：Apache 项目、活跃的 JVM 项目、支持 Python/Ruby 等多语言 API、被 30 多家公司使用——Twitter 用于个性化与搜索、Flipboard 用于生成定制 feed、Weather Channel 与 WebMD 等）。它的编程模型是有向图上的数据流处理：用户定义一张拓扑（Topology），拓扑里的节点不断接收上游的元组、做一次处理、再发出新的元组。
五个构件（Storm Components，讲义原口径）：
- 元组（Tuple）：元素的有序列表，是 Storm 里最小的数据单元。例：<tweeter, tweet>、<"Miley Cyrus", "Hey! Here's my new song!">、<URL, clicker-IP, date, time>，如 <coursera.org, 101.102.103.104, 4/4, 10:35:40>。Tuple 的字段有名字（fields），这是后面 fields grouping 能存在的前提。
- 流（Stream）：元组的序列，在数量上可能是无界的（potentially unbounded）。例：<"Miley Cyrus", "...">, <"Justin Bieber", "...">, <"Rolling Stones", "...">, ...。
- Spout：Storm 里流的源头，通常从爬虫、消息队列或数据库读取数据并 emit 元组。
- Bolt：Storm 里处理输入流并输出新流的实体。讲义列出 bolt 的常见”口味”：Filter（只转发满足条件的元组）、Joins（收到 A、B 两条流时输出所有满足条件的 (A,B) 对）、Apply/Transform（对每个元组施加一个函数），以及许多其他操作。
- 拓扑（Topology）：spout 与 bolt 组成的有向图，对应一个 Storm “应用”。它是一个长期运行的作业（与 MapReduce 作业”跑完即止”相反），必要时可以包含环（cycles）。
- 并行度（Parallelism）：一个 bolt 由多个 task（进程/线程）组成，一条输入流在多个 task 之间被拆分，通常每个输入元组只送往其中一个 task，具体由分组策略（Grouping Strategy）决定。
直观解释（”它是什么？”）：Storm 的拓扑就像一条工厂流水线：spout 是进料口，bolt 是各个工位，tuple 是在传送带上流动的零件，grouping 决定了”零件如何分派给同一工位的多个工人”。与 MapReduce 的静态”map→reduce 两阶段”相比，Storm 的拓扑是任意有向图 + 长期运行：你可以让一条流经过 10 个工位，也可以让两条流在半路汇合（join），还可以让某个工位的输出绕回来（有环拓扑，用于迭代式算法）。
机制图解：一个典型的 Storm 拓扑（含多 task 的 bolt 与 grouping 分派）。

                          ┌──────────────────────────┐
                          │  Spout（数据源，1~N 个）  │  读 Kafka / 爬虫 / DB
                          └────────────┬─────────────┘
                                       │ Stream(tuple1, tuple2, tuple3, ...)
                     ┌─────────────────┼─────────────────┐
                     │  grouping 决定分派（例如 fields grouping）
                     v                 v                 v
               ┌──────────┐      ┌──────────┐      ┌──────────┐
               │ Bolt1#0  │      │ Bolt1#1  │      │ Bolt1#2  │   同一个 Bolt 的 3 个 task
               │ (filter) │      │ (filter) │      │ (filter) │   （并行度 / parallelism hint = 3）
               └────┬─────┘      └────┬─────┘      └────┬─────┘
                    │                 │                 │
                    └─────────────────┼─────────────────┘
                                      │ （例如 all grouping：广播）
                                      v
                               ┌─────────────┐
                               │   Bolt2     │  聚合 / 计数 / join
                               └──────┬──────┘
                                      v
                               ┌─────────────┐
                               │ 输出 Bolt    │  写数据库 / 发告警 / 推送
                               └─────────────┘

   注意：拓扑是【长期运行】的；Spout 不断 emit，Bolt 不断处理——没有"作业结束"这个时刻。

分组策略（Stream Grouping）——本节最容易被考到的细节，讲义明确点出”三种最流行”，其余为完整语义（补充说明）：

Grouping	分派规则	语义保证	典型用途
Shuffle Grouping	轮询（round-robin）地在 bolt 的各个 task 之间均匀分发	每个 task 拿到大致相同数量的元组；同一 key 不保证落到同一 task	无状态、可任意并行的处理（解析、过滤、格式转换）
Fields Grouping	按元组的一个字段子集做哈希分区（讲义例：推特用户名首字符 `[A-H,a-h,0-3]` → task1，`[I-Q,i-q,4-6]` → task2，`[R-Z,r-z,7-9]` → task3）	相同字段值的元组必然落到同一个 task	有状态聚合：按用户/URL/会话计数、去重、top-K —— 这是”分布式 group by”的分区键
All Grouping	广播：bolt 的每个 task 都收到全部元组	每个 task 看到完整流（代价是 N 倍流量）	全局视图：全局阈值判断、全局配置/黑名单下发、需要与全量数据比对的 join
Global Grouping（补充）	整个流只送往 bolt 的 task 0	全局单点（可能成为瓶颈）	需要全局定序或全局唯一决策的场景
Direct Grouping（补充）	由上游代码显式指定目标 task	完全自定义路由	需要自己实现分区/路由逻辑（如按一致性哈希）
Local or Shuffle（补充）	优先发往同一 worker 进程内的 task，否则退化为 shuffle	减少跨进程/跨机网络开销	降低 shuffle 成本的通用优化

必须讲透的两点：(a) fields grouping 的本质是「保证同一 key 到同一 bolt」，因此它是在 Storm 里实现有状态聚合（分布式 group by）的唯一正确选择——若对计数类 bolt 用 shuffle grouping，同一个 key 会被分散到多个 task，每个 task 只看到局部计数，结果错了而且错得很隐蔽；(b) all grouping 的本质是「广播」，每条元组被复制 N 份（N = 并行度），只该用在「每个 task 都必须看到全量数据」的场景（全局阈值判断、配置/黑名单下发），误用会让网络与 CPU 开销随并行度线性膨胀。

关键假设与系统模型：Storm 的模型是异步消息传递 + 可能丢失/重复的可靠层：tuple 在网络中可能延迟、可能因为节点崩溃而丢失、也可能因为重放而重复。Storm 的容错设计因此必须显式处理”未确认“与”重复“两种情形，这就是 21.2.4 的 anchoring / acker / 超时重放要解决的问题。

21.2.4 Storm 的架构与可靠性机制：Nimbus / Supervisor / ZooKeeper / Anchoring / Acker

集群架构（Storm Cluster，讲义原口径）：
- Nimbus（主控/协调者，类似 master）：运行在 master 节点上的守护进程，负责把代码分发到集群各处、把 task 分配给机器、监控机器故障。它不参与数据的处理路径（这一点与 Hadoop 1.x 的 JobTracker 不同，JobTracker 还管任务调度与进度），因此 Nimbus 不是数据处理的热点。
- Supervisor（工作节点守护进程）：运行在每台工作机上，监听分派给本机的任务，并运行 Executor（executor 内部包含一组 task）。
- Worker Process / Executor / Task 的层级：一个 worker 进程（JVM）里可以跑多个 executor（线程），一个 executor 里可以跑多个 task。并行度（parallelism hint）就是 task 的数量，它是 Storm 伸缩性的旋钮：提高 parallelism hint 就是给某个 bolt 增加 task 数。
- ZooKeeper（协调服务）：协调 Nimbus 与 Supervisor 之间的通信，Nimbus 与 Supervisor 的全部状态都存放在 ZooKeeper 里。这个设计让 Nimbus 变成”无状态”的：Nimbus 崩溃后重启，从 ZooKeeper 读回拓扑与分配信息即可继续工作（对照 Lecture 15-B 的 Paxos：ZooKeeper 内部用 Zab/共识保证这些元数据的高可用）。

   ┌──────────────────────────────────────────────────────────────────────┐
   │                            Nimbus (master)                            │
   │   分发代码 / 分配 task / 监控机器故障        无数据处理路径           │
   └───────────────┬──────────────────────────────────────────────────────┘
                   │  拓扑提交 (job submission)
         ┌─────────┴──────────┐
         │  ZooKeeper Cluster │  ← Nimbus 与 Supervisor 的全部状态、心跳、
         │  (协调与故障恢复)   │     分配信息（元数据的强一致存储）
         └─────────┬──────────┘
                   │
     ┌─────────────┼──────────────┬───────────────┐
     v             v              v               v
 ┌────────┐   ┌────────┐     ┌────────┐      ┌────────┐
 │ W1     │   │ W2     │     │ W3     │      │ W4     │    Worker 节点（服务器）
 │Supervi-│   │Supervi-│     │Supervi-│      │Supervi-│    ← 监听本机任务
 │sor     │   │sor     │     │sor     │      │sor     │
 │ ┌────┐ │   │ ┌────┐ │     │ ┌────┐ │      │ ┌────┐ │
 │ │Exec│ │   │ │Exec│ │     │ │Exec│ │      │ │Exec│ │    Executor = 线程
 │ │task│ │   │ │task│ │     │ │task│ │      │ │task│ │    Task = 并行度单元
 │ └────┘ │   │ └────┘ │     │ └────┘ │      │ └────┘ │
 └────────┘   └────────┘     └────────┘      └────────┘

可靠性的核心问题：一个 spout tuple 发出后，会派生出一棵元组树（tuple tree）：spout 发出的 tuple 被 bolt1 处理，bolt1 可能发出 1 个或多个新 tuple；每个新 tuple 又被下游 bolt 处理、再发出新 tuple……只有当整棵树上的每个 tuple 都被成功处理，这条记录才算处理完毕。问题是：谁来判断”整棵树完成了”？ 如果让 spout 自己维护整棵树的状态，它就要记录每个后代 tuple 的信息，内存开销随树的规模无界增长（一棵树的 tuple 数可能是几千个）。Storm 的答案是 acker + XOR 校验值。
三个关键机制（讲义原口径 + 展开）：
1. 锚定（Anchoring）：emit(tuple, output) 中的第一个参数是输入元组（anchor），即”把输出锚定到一个或多个输入元组上“。锚定建立了 tuple tree 的父子边，它有两个作用：(a) 告诉 acker”新 tuple 属于哪些树”；(b) 失败传播——一个 tuple 的失败会导致一个或多个 tuple 被重放（讲义原文：failure of one tuple causes one or more tuples to be replayed）。锚定到多个输入元组时，新 tuple 同时属于多棵树（多流 join 场景）。
2. Acker（确认器）：每个 spout tuple 对应一个 acker task，它维护该树的校验值（ack value），用异或（XOR）来汇总”创建”与”完成”两类事件。
3. 超时与重放：讲义对失败的定义是：当一个元组及其派生出的元组图在指定超时时间内没有被完整处理时，该元组被视为失败。失败后由 spout 重放（replay）该 tuple（重放整棵树）。此外 Fail(tuple) 允许 bolt 立刻失败整棵树——讲义的原话是”如果数据库等抛出异常，立刻让处于拓扑树根部的 spout tuple 失败“（不必等超时）。
为什么用 XOR？（讲义只写了机制，这里补齐理由）：XOR 的三个代数性质恰好匹配分布式确认：(1) 交换律与结合律——同一棵树的不同分支由不同 bolt 并行处理，ack 到达 acker 的顺序完全任意，XOR 让校验值与到达顺序无关；(2) 自逆性 $x\oplus x=0$——「创建」与「完成」这对事件天然互相抵消，不需要查找表（加法会累加出 $2x$，乘法无法表达「取消」）；(3) 状态极小——每个 spout tuple 只需 8 字节，与树的规模无关（显式维护 pending 集合需要 $O(\text{树中 tuple 数})$）。
机制图解：拓扑、tuple tree 与 XOR 校验值演算（本章最重要的图之一）。设 spout tuple $T_0$ 派生出两个子 tuple $T_1, T_2$，$T_1$ 再派生出 $T_3$，$T_2$ 再派生出 $T_4$；为了便于阅读，图中把 64 位 id 截断成 16 位十六进制。

   tuple tree（锚定关系：箭头 = "anchored on"）
   ┌────────────────────────────────────────────────────────────────┐
   │                     ┌───────────────────────┐                  │
   │                     │  T0  (spout, id=A001) │  ← 树的根          │
   │                     └───────────┬───────────┘                  │
   │                    anchored ┌───┴───┐ anchored                 │
   │                             v       v                          │
   │          ┌──────────────────────┐  ┌──────────────────────┐    │
   │          │ T1 (bolt1, id=B002)  │  │ T2 (bolt1, id=C003)  │    │
   │          └──────────┬───────────┘  └───────────┬──────────┘    │
   │              anchored│                  anchored│              │
   │                      v                          v              │
   │          ┌──────────────────────┐  ┌──────────────────────┐    │
   │          │ T3 (bolt2, id=D004)  │  │ T4 (bolt2, id=E005)  │    │
   │          └──────────────────────┘  └──────────────────────┘    │
   └────────────────────────────────────────────────────────────────┘
   （一个输入 tuple 可以锚定出多个输出 tuple ⇒ 树会分叉；每个分支被独立并行处理）

   acker 中该树的 XOR 校验值 V 的完整演算（事件到达顺序任意，结果相同）：
   ┌────────────────────────────────────────┬───────────┬───────────┬──────────────────┐
   │ 事件                                    │ 操作数     │ V（旧）    │ V（新）           │
   ├────────────────────────────────────────┼───────────┼───────────┼──────────────────┤
   │ create T0 （spout 发出根 tuple）        │ ^ A001    │ 0000      │ A001             │
   │ create T1 （bolt1 发出，锚定在 T0 上）  │ ^ B002    │ A001      │ 1003             │
   │ create T2 （bolt1 发出，锚定在 T0 上）  │ ^ C003    │ 1003      │ D000             │
   │ ack    T0 （bolt1 处理完 T0）           │ ^ A001    │ D000      │ 7001             │
   │ create T3 （bolt2 发出，锚定在 T1 上）  │ ^ D004    │ 7001      │ A005             │
   │ ack    T1 （bolt2 处理完 T1）           │ ^ B002    │ A005      │ 1007             │
   │ create T4 （bolt2 发出，锚定在 T2 上）  │ ^ E005    │ 1007      │ F002             │
   │ ack    T2 （bolt2 处理完 T2）           │ ^ C003    │ F002      │ 3001             │
   │ ack    T3 （sink 处理完 T3）            │ ^ D004    │ 3001      │ E005             │
   │ ack    T4 （sink 处理完 T4）            │ ^ E005    │ E005      │ 0000  ✓ 归零！    │
   └────────────────────────────────────────┴───────────┴───────────┴──────────────────┘
   V = 0  ⟹  树中不再有"已创建但未确认"的 tuple  ⟹  整棵树处理完毕 ⟹ 通知 spout：成功！

   反例（失败检测）：
   create T0 ^A001 = A001     create T1 ^B002 = 1003     ... 假如此后 bolt2 崩溃，
   ack T1 与 ack T3 永远不会到达  ⟹  V ≠ 0  ⟹ 超时后 spout 重放 T0 整棵树

OutputCollector API（讲义原口径）：emit(tuple, output)（发出一个输出元组，可选地锚定到输入元组，第一个参数是 anchor）、ack(tuple)（确认”我处理完这个元组了”）、fail(tuple)（因为数据库异常等原因，立即让元组树根部的 spout tuple 失败）。讲义特别警告：必须记得对每个 tuple 调用 ack/fail——每个 tuple 都会占用内存，忘记确认会导致内存泄漏。
Storm 的语义边界：默认语义是 at-least-once（至少一次）：超时未确认 ⇒ 重放 ⇒ 已处理过的那部分会被重复处理，因此结果可能出现重复。要做到 exactly-once，Storm 需要 Trident（基于事务性拓扑的高层 API）——D-Streams 论文对它的评价很直接：Trident 通过”把状态保存在一个复制数据库中、按批提交更新”来简化编程接口，但代价是基线的处理成本上升（需要把更新事务性地复制到网络上）。这也解释了为什么 Spark Streaming 与 Flink 后来走了一条不同的路（见 21.2.6、21.2.7）。
性能优化的演进（讲义补充材料）：Twitter Heron 修复了 Storm acking 机制的低效之处并使用背压——拥塞的下游分级要求上游分级/spout 减速，三条路径是 TCP 背压（TCP 窗口机制）、Spout 背压（停止从上游 spout 读）、逐级背压（按 stage 传播），推荐组合为「Spout + TCP」或「Stage-by-stage + TCP」，讲义结论是 Heron 的吞吐轻松超过 Storm。根因值得记住：tuple 级 ack 的记账开销与树的 tuple 数同阶，记账流量可能吃掉可观比例的带宽（21.5 量化）。

21.2.5 Apache Spark：RDD、血缘（Lineage）与 DAG 调度

动机（必须讲透）：Spark 一讲的出发点是反问”做大数据分析，MapReduce 还不够好吗？“（Isn’t MapReduce good enough?）。MapReduce 擅长在大量廉价节点上简化批处理，用”任务重执行 + 中间结果落盘”换取极强容错（Lecture 5）；它昂贵的是迭代式（iterative）与交互式（interactive）应用，根因是为容错而反复写磁盘（讲义原话 expensive save to disk for fault tolerance），进而缺乏高效的数据共享——于是业界为每种编程模型各造轮子（Pregel、HaLoop），每个都要重新解决容错与数据共享。Spark 的答案是：用一个数据抽象（RDD）+ 一个容错机制（血缘）同时支撑批处理、流处理、图处理与机器学习（讲义：RDD 允许统一流处理、图处理、机器学习等不同编程模型）。
RDD（Resilient Distributed Dataset，弹性分布式数据集）的三条性质（讲义原口径）：
1. 不可变（immutable）、分区（partitioned）的记录集合。不可变意味着没有原地更新：所有操作都是”由旧 RDD 生成新 RDD”，因此数据一旦生成就不会被并发修改（对照 Lecture 21 的并发控制：不可变数据天然免锁、天然可重算）。
2. 通过粗粒度（coarse-grained）的转换构建：map、filter、join、groupBy、reduceByKey……”粗粒度”的含义是对全部分区施加同一个操作（而不是对某个元素做定点修改）。这个性质是血缘容错的前提：因为每个分区的生成规则都一样，所以”重算一个分区”就等于”用同一段代码 + 同一份父分区再跑一次”。
3. 可以被缓存（cache/persist）在内存中供复用——这是 Spark 比 MapReduce 快的核心。讲义给出的图景是：HDFS 读入 → RDD → map → RDD → reduce → RDD，中间结果可以 cache 留在内存里，下一次迭代直接读内存，而 MapReduce 必须回到 HDFS。
容错机制：血缘（Lineage）。讲义的原话极其精炼：”日志记录（log）施加在分区数据集上的粗粒度操作；如果发生失败，就简单地重算丢失的分区；没有失败就没有代价（No cost if no failure）“。展开成三条：
- 血缘是什么：每个 RDD 记录”我是由哪个（哪些）父 RDD、经过什么转换、怎么分区得到的”。血缘是一张逐分区（partition-level）的依赖图，而不是数据本身。
- 恢复怎么做：某个节点挂了，它的 RDD 分区丢失，调度器只重算这些分区：重新读入它们对应的源分区，按血缘依次重放转换函数。不需要复制数据，不需要回滚整个作业。
- 成本模型：稳态下零开销（不复制、不落盘），代价全部推迟到故障发生时——恢复时间 ∝ 血缘链长度。因此血缘长到一定程度必须截断：D-Streams 论文把这一优化叫 lineage cutoff——”RDD 被检查点（checkpoint）之后，调度器就忘掉它的血缘，避免其状态无限增长”；Spark Streaming 也”周期性地对状态 RDD 做检查点（例如每五个 RDD 复制一次），以防无限重算”。
必须对比：MapReduce 的 re-execution vs Spark 的 lineage。两者都是”重新计算代替恢复状态”（这与 Lecture 5 的哲学一脉相承），但粒度与代价不同：

维度	MapReduce 的任务重执行	Spark 的 RDD 血缘重算
重算粒度	Task 级（一个 map/reduce task 失败 → 重跑该 task）	分区级（一个 RDD 分区丢失 → 只重算该分区的血缘）
中间结果	必须落盘（map 输出写本地盘，跨阶段走 HDFS）	默认留在内存（`cache`/`persist`），只在必要时 spill
稳态开销	高：shuffle 与中间结果都要写磁盘、跨机复制	低：内存复用，无失败则零额外开销
恢复速度	慢：受限于磁盘 IO 与被重算的阶段数	快：分区级并行重算（D-Streams 论文实测恢复延迟常 <1 秒）
恢复的假设	输入在 HDFS 上可靠 + 任务确定性	同左，额外要求转换是确定性的（否则重算结果不等于原值）
迭代/交互	差（每轮都要读写 HDFS）	好（缓存中间 RDD 供下轮复用；还能接交互式 shell 做 ad-hoc 查询）

窄依赖 vs 宽依赖（Narrow vs Wide Dependency）——stage 划分的根据：
- 窄依赖（narrow dependency）：一个父分区最多被一个子分区使用。例如 map、filter、flatMap、union。特点：(a) 可以在同一个 task 内流水线执行（父分区算完直接喂给子分区，不需要物化）；(b) 恢复代价低（丢失一个子分区只需重算对应的一个/几个父分区，甚至可以并行重算多个丢失分区）；(c) 不需要跨网络传输。
- 宽依赖（wide dependency / shuffle dependency）：一个子分区依赖多个父分区。例如 groupByKey、reduceByKey、join、repartition。特点：(a) 必须先让所有父分区产出，再按 key shuffle 到目标分区（网络 + 磁盘开销）；(b) 恢复代价高（丢失一个子分区可能要重算全部父分区，甚至级联重算）；(c) 它是 stage 的切分边界。
- DAG 调度：Spark 从 action 出发反向遍历依赖图，遇到宽依赖就切一刀：刀与刀之间是一个 stage，stage 内部全是窄依赖，可以流水线执行；stage 之间必须等 shuffle 完成。这个”宽依赖处切分 stage，stage 内流水线“的规则，是 Spark 既快（少落盘）又能容错（血缘清晰）的关键。

   ① 血缘 DAG 的分区级视图（每个椭圆 = 一个 RDD，椭圆内的小圈 = 分区）
                    窄依赖 (narrow)                    宽依赖 (wide / shuffle)
        srcRDD          Stage 1（可流水线执行）        Stage 2
     ┌──────────┐      ┌──────────┐   ┌──────────┐   ┌──────────┐    ┌──────────┐
     │ ○ ○ ○ ○  │─map─>│ ○ ○ ○ ○  │─map─>│ ○ ○ ○ ○  │═reduceByKey═>│ ○ ○ ○ ○  │─map─> 输出
     │ p0 p1 p2 p3     │ p0 p1 p2 p3     │ p0 p1 p2 p3   (shuffle)  │ p0 p1 p2 p3
     └──────────┘      └──────────┘   └──────────┘   └──────────┘    └──────────┘
        父分区 i  ──────> 子分区 i        父分区 i ──> 子分区 i        多个父分区 ──┐
     （一对一/一对少：窄）               （一对一/一对少：窄）            多对一 ────┴─> 一个子分区
                                                                        （需 shuffle，恢复代价高）
   ② 图中文字版 DAG（word count）
        srcRDD ──flatMap──> words ──map──> pairs ══reduceByKey══> counts ──map──> outRDD
        └──────────── Stage 1（pipeline，无 shuffle）────────┘   ↑ 宽依赖边界   └─ Stage 2 ─┘
                                                                  （stage 划分点）

   ③ 失败恢复的差异（同一个故障，两种依赖的代价完全不同）
     窄依赖下丢失 j 分区：只需重算 p_j 链 ⇒ 1 个任务（可并行）
     宽依赖下丢失 j 分区：需要全部父分区重算 + 重新 shuffle ⇒ 可能触发 stage 级重算

关键假设与系统模型（血缘恢复的正确性前提，21.3.2 会正式证明）：
- (a) 确定性的转换（determinism）：同一个父分区 + 同一段函数 ⇒ 同一个输出分区。任何非确定性（random()、System.currentTimeMillis()、遍历哈希表的顺序、未排序的并发写入）都会破坏这条等式。工程做法是把非确定性提升为显式输入（例如随机种子、时间戳作为 RDD 的一个字段）。
- (b) 源数据可靠且可重放：HDFS 副本、可回放的 Kafka offset，或一个检查点（checkpoint）。
- (c) 血缘被完整记录：包括分区映射关系（谁依赖谁）。这是 RDD 的 dependencies 字段的职责。
- 在这三条之上，重建是”逐字节等价”的，因此血缘恢复不但能修复数据，还能支撑推测执行（speculative execution）：把慢任务的副本跑到另一台机器上，谁先完成用谁的结果——这在记录级（record-at-a-time）的流系统里极其困难（要复制并追赶一个有状态节点的状态），在 Spark/D-Streams 里却是”再跑一遍同一个确定性任务”这么简单。
Generality 与 Partitioning（讲义延伸）：RDD 的通用性体现在图处理上——GraphX 用「聚集-应用-散布（GAS）」表达图算法（group-by 聚集邻居 → gather 汇总 → apply 更新 → scatter 沿边散布 → join 生成新三元组），结论是图算法不需要新系统，只需要「能在内存里反复复用同一份分区数据」的抽象。讲义另用 PageRank 说明分区的重要性：算法每轮都要 join Links 与 Ranks，若两者分区方式一致就无需 shuffle（否则每轮迭代都跨网络）——「一次分区、多轮复用」是 Spark 里最重要的性能优化之一。
RDD 上的操作分类：转换（transformations）：filter/join/map/group-by（惰性，只建血缘）；动作（actions）：count/print（触发 DAG 执行）；控制：partitioning（分区方式）、persistence（是否缓存）。

21.2.6 Spark Streaming 与 DStream：微批（Micro-batch）模型

定义与目的：Spark Streaming 把流处理归约成批处理：把输入流按固定时间间隔（讲义与 D-Streams 论文的目标是亚秒级，如 0.5 秒；工程中常用 1 秒）切成一系列小批，每个批是一个 RDD，因此
\[\textbf{DStream} = \text{RDD 的序列（a sequence of RDDs）}\]
这就是 D-Streams（Discretized Streams，离散化流） 名字的来源。它的设计赌注是：用固定的批间隔（秒级）换来”完全确定性的批计算 + 成熟的批容错机制”——D-Streams 论文明确指出，目标应用（社交趋势、垃圾邮件检测、集群监控、网络入侵检测）的时间尺度远大于 1 秒，因此秒级延迟是可以接受的，而”无复制、恢复快”的收益远大于延迟代价；论文同时明确排除需要几百毫秒以内延迟的应用（如高频交易）。
直观解释（”它是什么？”）：把流处理想成用秒表切香肠：不管香肠（数据流）多长，你每秒切一刀，每一段都当作一个”小数据集”去算。好处是算每一段的方法和批处理完全一样（同一套 RDD API、同一套血缘容错、同一套调度器），坏处是结果最快也要等到下一刀落下——这就是微批的延迟下限。
机制图解：微批、DStream 与窗口。

   无界输入流:  ▒▓▒▓▒▓▒▓▒▓▒▓▒▓▒▓▒▓▒▓▒▓▒▓▒▓▒▓▒▓▒▓▒▓▒▓▒▓▒▓▒▓▒▓▒▓▒▓   (永不结束)
   批间隔 (batch interval) = 1s
                |-- t1 --|-- t2 --|-- t3 --|-- t4 --|-- t5 --|-- t6 --|-- t7 --|
   DStream:      [ RDD1 ] [ RDD2 ] [ RDD3 ] [ RDD4 ] [ RDD5 ] [ RDD6 ] [ RDD7 ]
                  ↑ 每个批 = 一个不可变、可分区、可缓存的 RDD（继承 Spark 的全部容错机制）

   窗口操作 window(windowLength=3s, slideInterval=1s)：
                [────── RDD1..RDD3 ──────]
                        [────── RDD2..RDD4 ──────]
                                [────── RDD3..RDD5 ──────]
   增量聚合 reduceByKeyAndWindow(3s, 1s, invReduceFunc)：
        新窗口 = 旧窗口 + 进入的批 - 滑出的批      （需要聚合函数有逆元，如 count/sum 的减法）
        无逆元版本 = 每轮把窗口内 3 个批重新合并     （有逆元版本省掉重复求和）

   有状态操作：
        updateStateByKey / mapWithState：状态 RDD(t) = f(状态 RDD(t-1), 批(t))
        ⇒ 状态本身也是 RDD ⇒ 也靠血缘（+ 周期检查点）来容错，而不是靠"热备复制"

算子清单（D-Streams 论文口径）：无状态转换（与批处理一致：map/filter/reduceByKey/join，输出只依赖同一区间的父 RDD）；窗口算子 window(windowLength, slideInterval)（例如 sentences.window("5s") 得到 [0,5), [1,6), … 的 RDD 序列）；增量聚合 reduceByWindow(assoc) 与带逆函数的 reduceByWindow(assoc, invAssoc)——论文的对照实验说明了差别：两种版本都只对每个 interval 计数一次，但带逆函数的版本用「加新批、减旧批」避免了每轮重复求和；状态跟踪 track(init, update, timeout)（论文的视频会话例子），对应现代 API 的 updateStateByKey/mapWithState；输出算子 save 与 foreachRDD（端到端语义的最后一环，见 21.3.3）。
容错：血缘 + 可靠重放的输入 + 检查点。(1) 血缘重算：丢失的 RDD 分区按血缘重建（分区级、可并行，这是它远快于「上游备份」的原因）。(2) 输入必须可重放：论文的硬约束是——输入数据先被复制到两个 worker 才算接收成功，失败时客户端把未确认数据重发（对应 Kafka 场景就是保存并重放 offset）。(3) 检查点：保存 DStream 计算图 与有状态算子的状态 RDD，并配套 lineage cutoff（忘掉已检查点的血缘，避免状态无界增长）。(4) 再加上输出端的幂等/事务性，才构成端到端 exactly-once。
微批的局限（最容易被追问的点）：(1) 延迟下限 = 批间隔——哪怕集群空闲、只来一条记录，端到端延迟也至少是”等这一批结束”（1 秒批 ⇒ 秒级下限，0.5 秒 ⇒ 亚秒级），Spark Streaming 在原理上做不到毫秒级；(2) 批间隔不是越小越好——每批的调度与提交开销占比上升，吞吐下降（论文用”1 秒延迟目标配 500 ms 批间隔、2 秒目标配 1 秒批间隔”说明这一点）；(3) 批边界上的对齐会让一个慢的尾任务拖住整批（论文用 timestep pipelining 缓解：允许下一时间步的任务在当前步未结束时提交）。因此出现两条改进路线：结构化流（Structured Streaming）（把流抽象成无界表上的增量查询，用 watermark 处理迟到、用幂等/事务 sink 保证端到端 exactly-once）与 Flink（放弃微批，走真流 + barrier checkpoint，见 21.2.7）。

21.2.7 Flink 与 Chandy-Lamport barrier checkpoint：Lecture 12 的快照算法就在流处理的引擎里

定义与目的：Flink 用 barrier（屏障）+ 对齐（alignment）+ 状态快照实现 exactly-once，而barrier 就是 Lecture 12 讲的 Chandy-Lamport 快照算法里的 marker（标记）。这不是类比，而是同一个算法：
- Chandy-Lamport 的 marker ↔ Flink 的 barrier：由某个进程（Flink 里是 source 算子）在自己的输出通道上注入，随应用消息一起沿通道传播；接收方按通道 FIFO 顺序收到它，从而知道”这条 marker 之前的所有消息都属于快照之前”。
- Chandy-Lamport 的 进程状态（local state） ↔ Flink 的算子状态（operator state / keyed state）。
- Chandy-Lamport 的 通道状态（channel state） ↔ Flink 的输入缓冲区（in-flight records）：在”barrier 之后的记录不能进入快照前状态”的规则下，Flink 采用”停读该通道“的方式替代”把通道内容记进快照”（这正是对齐式 checkpoint）。
- Chandy-Lamport 的 快照收集（全局一致割） ↔ Flink 的 checkpoint 完成（所有算子上报、JobManager 确认）；只有完成的检查点才能用于恢复。
- 两处重要的工程扩展（讲义之外的必要补充）：(a) barrier 携带 checkpoint id（$n$），因此支持多代检查点并发（不必停止注入新数据、不必全局暂停）；(b) 快照要与外部输出协同，因此 Flink 的 sink 用两阶段提交（2PC）：checkpoint 完成前输出处于”预提交”状态，完成后才真正提交——这让”状态 + 输出”一起回滚，达到端到端 exactly-once。
直观解释：把作业想成一条多车道高速路，每个算子是一个收费站。要拍「所有车在同一时刻的位置」这张照片是不可能的，于是 Chandy-Lamport 的做法是：每个入口同时放下一道红色横杆（barrier），横杆以与车流相同的速度推进；每个收费站等到各条车道的横杆都到了（对齐），才记录「此刻本站的账目（状态）」，再把横杆传给下一站。所有收费站都记录完，就得到一张一致的全局照片。代价也一目了然：先到横杆的车道必须靠边等——这就是对齐式 checkpoint 的停顿。
机制图解：barrier 对齐过程（明确标注与 Lecture 12 的 Chandy-Lamport marker 的对应关系）。

   Lecture 12 (Chandy-Lamport):         Flink（现代流处理引擎）:
   ─────────────────────────────        ─────────────────────────────────────────
   进程 P_i 在输出通道上发 marker   ⇔   source 算子在每条输出通道上注入 barrier
   marker 与普通消息共用通道、FIFO  ⇔   barrier 与数据记录在同一通道上按序流动
   每个通道收齐 marker 后记通道状态 ⇔   收到某通道 barrier 后【停读该通道】(对齐)
   进程收齐所有输入通道的 marker    ⇔   算子收齐所有输入通道的 barrier ⇒ 对齐完成
   然后记录本地状态                 ⇔   然后 snapshot 本地状态（keyed state / operator state）
   快照集合 = 全局一致割            ⇔   checkpoint 完成 = 所有算子 ack + JobManager 确认

   源 S1 ──[a1][a2]──[M]──[a4][a5]───────────────────────────────┐
   源 S2 ──[b1][b2][b3][b4]──[M]──[b6]───────────────────────────┤   （M = barrier，
                                                                  │    n = checkpoint id）
                         两条通道汇入同一个有状态算子 op2          │
                                                                  v
   op2 的输入缓冲区:
        通道A:  [a1 a2] |            (停读！后续 a3 a4 a5... 缓存在缓冲区内)
        通道B:  [b1 b2 b3 b4] |      (M 到达) ⇒ 两条通道的 barrier 都到齐 ⇒ 对齐完成
                                 │
                                 ├─ ① snapshot op2 的状态（= Chandy-Lamport 的"记录本地状态"）
                                 ├─ ② 把 M 广播给下游所有输出通道（继续传播快照）
                                 └─ ③ 恢复读取通道 A，处理对齐期间缓存的记录
                                       （这些记录属于【本次快照之后】的数据，因此放在快照之后处理）

   时间轴视图（对齐造成的停顿一目了然）:
        step:   1    2    3    4    5    6    7
   通道A:       .    .    |    x    x    x    .      . = 已处理  | = barrier  x = 缓存等待
   通道B:       .    .    .    .    .    |    .      对齐完成的时刻 = max(各通道 barrier 到达时刻)

恰好一次是怎么闭环的：Flink 的端到端 exactly-once 由三个齿轮咬合——(1) 状态可回滚（从最近一个已完成的 checkpoint 恢复，未完成的检查点直接丢弃）；(2) 输入可重放（把 source 的读取位置重置到 barrier 所标记的偏移，如 Kafka offset）；(3) 输出可幂等/事务（sink 用 2PC：只有检查点完成时预提交的输出才真正提交，因此”回滚 + 重放”不会在外部系统留下重复）。详见算法 21.3.4。
对齐式（aligned）vs 非对齐式（unaligned）checkpoint——一个必须知道的现代权衡：

	对齐式 checkpoint（aligned，Flink 1.x 默认）	非对齐式 checkpoint（unaligned，Flink 1.11+）
barrier 的处理	barrier 到达某通道后停读该通道，等其它通道的 barrier	barrier 可以超过队列中已有的记录（”插队”）
对齐期间的记录	缓存在内存/磁盘，快照后处理	被当作通道状态写进快照（先落盘，不等对齐）
停顿	有：背压越重，对齐越慢，延迟尖峰越大	几乎无停顿：不做对齐等待
快照大小	小（只含算子状态）	大（含 in-flight 数据）
适用场景	正常负载、低背压	持续高背压、超大状态、延迟敏感

对比表：Storm vs Spark Streaming vs Flink vs Kafka Streams（本章要求的核心对照表）：

维度	Storm（+Trident）	Spark Streaming（D-Streams）	Flink	Kafka Streams
处理模型	逐记录（record-at-a-time），长期运行的有状态算子组成拓扑	微批（micro-batch）：DStream = RDD 的序列，按批间隔切分	真正的逐记录流（dataflow + 有状态算子），批是流的特例	逐记录，库（library）而非集群（嵌在应用里，用 Kafka 做状态存储/协调）
延迟	毫秒级（最小延迟最好）	秒级（下限 = 批间隔；论文目标 0.5–2s）	毫秒~亚秒级（barrier 机制不引入批等待）	毫秒~秒级（取决于 commit interval 与缓存）
投递语义	at-least-once（默认）；Trident 可达 exactly-once（事务性拓扑 + 复制数据库状态）	微批天然提供”每批原子”，配合可靠重放输入 + 幂等/事务输出可达 exactly-once	原生 exactly-once（barrier checkpoint + 2PC sink）	exactly-once（Kafka 事务：把状态更新与 offset 提交放进同一事务）
容错机制	tuple 级 ack（XOR 校验值）+ 超时重放 + 锚定；失败重放整棵树	RDD 血缘 + 分区级重算 + 检查点（lineage cutoff）；上游备份的替代方案：并行恢复	Chandy-Lamport barrier checkpoint（对齐/非对齐）+ 状态后端快照 + 2PC 输出	Kafka 事务日志（changelog topic）：state 存本地 + 变更写入 changelog，故障后重建
状态管理	用户代码自理（Trident 用外部复制数据库）；ack 状态在 acker 内存	状态是 RDD（可缓存/可血缘重算），需周期性 checkpoint	一等的 keyed state / operator state，可插拔状态后端（内存 / RocksDB），支持大状态与增量检查点	一等的本地状态 + changelog topic（Kafka 就是它的持久层）
背压	Storm 原生较弱（acking 开销 + 队列堆积）；Heron 用 TCP/spout/逐级背压改进	微批天然背压（批处理不完就推迟下一批）	网络层背压 + 对齐停顿（可用 unaligned 缓解）	Kafka 消费端背压由 `max.poll.records` / 拉取节奏控制
吞吐量级	论文实测 1000 字节记录下约 2× 慢于 Spark Streaming；100 字节记录下差距更大	论文实测 100 节点上 60M records/s（6 GB/s）亚秒级延迟；Grep 场景 670K records/s/node vs Storm 115K	高（逐记录 + 高效网络栈 + 状态后端优化）	高（单应用内，无独立集群开销）
与批处理统一	无（拓扑与批处理模型完全不同）	强：同一套 RDD API、同一份代码可跑批/流/交互查询	强：DataStream/DataSet 统一，Table/SQL 同一引擎	弱（只做流；批用 Kafka Connect/其它）

这张表的读法

这张表的读法：“语义强度”与”延迟”是两条独立的轴。Storm 延迟最低但语义最弱；Spark Streaming 语义可以做到 exactly-once，代价是延迟被批间隔钉死；Flink 用 Chandy-Lamport 快照同时要到了低延迟与 exactly-once；Kafka Streams 则用”把状态放进 Kafka 事务”这条捷径换来了轻量部署。它们用的容错思想都是”重算 + 快照”，没有一个是”热备复制”——这正是本章黄金法则的第一半。

21.2.8 集群调度的演进：从单体调度到共享状态调度

为什么需要调度（讲义三段式）：(1) 有大量「任务」需要安排（单核 OS 上的进程、一个 Hadoop 作业的任务、多个 Hadoop 作业的任务）；(2) 资源有限且多维（处理器、内存是「有争议的」，磁盘与网络相对不紧张——讲义原文 less contentious）；(3) 两个目标：为任务/作业取得好的吞吐量或响应时间，以及资源的高利用率。讲义还点出关键视角：云用户（我的作业多快完成）与云提供商（集群利用率与总吞吐）的指标不同，二者天然冲突，而公平性就是把冲突「制度化」的工具。
单处理器调度的四个经典算法（讲义的数值例子贯穿始终：三个任务，长度分别为 10、5、3，到达时间分别为 0、6、8）：

算法	规则	讲义数值例的平均完成时间	特点
FIFO / FCFS	按到达顺序排队，处理器空闲就取队头	$(10+15+18)/3 = 43/3 \approx \mathbf{14.33}$	最简单；长作业在队头会阻塞后面的短作业（队头阻塞，必须等到 10 结束）
STF（Shortest Task First）	按运行时间从小到大排队（假设运行时间已知）	$(18+8+3)/3 = 29/3 \approx \mathbf{9.66}$	讲义明确：”STF 在所有调度方法中平均完成时间最短，它是最优的”；缺点是需要预知运行时间，且长作业可能饿死（讲义思考题：”如果任务 2 与任务 3 都在 t=1 到达会怎样？”——答案：它们会排在 10 之前，反过来饿死长任务）
优先级（Priority）	用用户提供的优先级代替”运行时间”作排序键	—	讲义指出 STF 是优先级调度的一个特例（优先级 = 任务长度的倒数）
轮转（Round-Robin）	用一个时间片（quantum）运行队头任务的一段，抢占时保存进程状态，然后把它放到队尾；切片过小则切换开销过大	—	适合交互式应用（用户要快速响应）；FIFO/STF 适合批处理（提交后走开，回来取结果）

Hadoop 的两大调度器（讲义重点）：
- Capacity Scheduler（容量调度器）：多个队列，每队列含多个作业，每队列被保证获得集群的一定比例（例：队列 1 得 80%、队列 2 得 20%，高优先级作业进队列 1）；队内通常用 FIFO，一个作业可占满本队列份额，任务不够时下一个作业可启动填补；管理员可配置软限制（保证占比）与可选的硬限制（最大占比）。弹性（elasticity）：空闲时可超额占用，但其它队列回到容量下限时须归还；关键限制——不允许抢占（不能中途停任务，只能等任务自然结束再回收；好处是没有重算浪费）。此外支持层级队列（子队列间可平均分享）并能考虑用户指定的内存需求。
- Fair Scheduler（公平调度器）：目标是所有作业获得相等份额——只有一个作业时独占集群，其它作业到达后每个作业获得相等的集群百分比（讲义例：每作业得到相同数量的 YARN 容器，一个容器 = 一个任务）；实现上把集群划分成池（pool）（通常一用户一池），资源在池间平均、池内可配公平或 FIFO；当某池有最小份额且长时间未满足时，从其它池抢占——杀掉正在运行的任务，之后可重启，讲义明确指出”这没问题，因为任务是幂等的“，且为减少浪费优先杀最近才启动的任务；还可限制每用户/每池的并发作业数与每池并发任务数。
- 为什么不用 STF（它明明最优）？ 讲义的答案：任务完成前很难知道预期运行时间。工程近似有两条：作业内把任务时长建模为与其输入数据量成正比；跨任务把某作业中一个任务的时长估计为该作业其它任务的平均（按输入规模加权）。这也是延迟调度与任务时长预测研究的出发点。
- 讲义留下的伏笔：以上都只涉及一种资源（处理器或内存），多维资源需求怎么办？ —— 这就是 DRF。
调度框架的五代架构（本章要求的对比，讲义只覆盖到两级调度的思想，其余为补充说明）：

  (1) 单体调度 (Monolithic)          (2) 静态分区 (Static Partitioning)
    ┌────────────────────┐              ┌─────────┬─────────┬─────────┐
    │  唯一调度器          │              │ Hadoop  │ Spark   │ MPI     │
    │  (Hadoop 1.x        │              │  40%    │  30%    │  30%    │
    │   JobTracker)       │              └─────────┴─────────┴─────────┘
    └─────────┬──────────┘              集群被硬切成互不借用的孤岛
              │ 所有框架都接进来           忙闲不均时无法互相借用资源
    ┌─────────┴──────────┐               ⇒ 整体利用率低（典型 20%~40%）
    │      整个集群        │
    └────────────────────┘
    问题: 只支持一种计算框架; 调度器状态与集群规模强耦合

  (3) 两级调度 (Two-level / Mesos)      (4) 共享状态调度 (Shared-state / Omega)
    ┌───────────────────────────┐         ┌────────┐ ┌────────┐ ┌────────┐
    │      Mesos Master          │         │Sched A │ │Sched B │ │Sched C │
    │  按 DRF 决定"给谁多少"      │         └───┬────┘ └───┬────┘ └───┬────┘
    └────┬───────────────┬───────┘             │ 并发读/写完整集群状态
   resource offer   resource offer             │ （乐观并发控制 + 冲突重试）
    (资源邀约)        (资源邀约)                 v
   ┌────v─────┐    ┌────v─────┐         ┌──────────────────────────┐
   │Framework │    │Framework │         │  共享的集群状态 (cell)     │
   │scheduler │    │scheduler │         └──────────────────────────┘
   │(Spark/   │    │(Hadoop/  │         优点: 调度吞吐高、可并行决策
   │ MPI/...) │    │ MPI/...) │         缺点: 冲突时需重试(乐观并发)
   └──────────┘    └──────────┘
   优点: 支持多框架、可扩展
   缺点: offer 先到先得, 局部最优 ≠ 全局最优

  (5) 集中式 + 全局最优 (Centralized: Borg / Kubernetes)
    ┌──────────────────────────────────────────────┐
    │  单一（或分片）大调度器 + 优先级 + 抢占 + 约束求解 │
    │  Kubernetes: filter(过滤不可行节点) → score(打分) │
    └───────────────────────┬──────────────────────┘
                            v
                  Pod / Task 的放置决策（可带亲和/反亲和、污点容忍、QoS）

架构对比表（讲义口径 + 补充说明）：

架构	代表系统	放置决策者	多框架	可扩展性	效率	主要缺陷
单体（Monolithic）	Hadoop 1.x JobTracker	唯一中央调度器	差	差（单点、随规模退化）	中	无法服务异构框架；单点瓶颈
静态分区	「每框架一套静态集群」的传统运维	各框架各管一块	名义支持（资源不能流动）	好	差（忙闲不均，利用率 20%–40%）	资源孤岛；运维成本随框架数上升
两级调度	Mesos（+ DRF）	master 决定 offer，框架决定接不接受	好	好	好（细粒度共享 + DRF 公平）	offer 先到先得；局部决策≠全局最优
共享状态	Omega（Google）	多个调度器并发读写完整集群状态	好	很好（无中心队列）	好	乐观并发：冲突需重试，实现复杂
集中式+全局最优	Borg、Kubernetes	单一（或分片）大调度器 + 优先级/抢占/约束	好	中~好（靠分片与缓存）	好（可做全局最优与抢占）	单点扩展压力；抢占有重算开销

资源邀约（Resource Offer）机制：Mesos 把「资源分配」拆成两次决策：(1) Master 决定 offer——按 DRF 决定「把多少资源提供给哪个框架」，offer 是一个资源向量（如「节点 A 上的 4 CPU、8 GB」）；(2) Framework scheduler 决定接受或拒绝——按自己的任务队列与放置约束决定启动哪些任务，不接受就拒绝或超时，资源被收回后再 offer 给别人。优点：支持多种计算框架（各自保留调度逻辑）、扩展性好（master 工作量与「框架数 × offer 频率」相关，而非任务数）、粒度可细可粗。缺点：(a) 先到先得——迟疑的框架会让资源空转；(b) 局部决策≠全局最优；(c) 不支持复杂全局约束（gang scheduling、跨任务亲和需要框架自己实现）。这正是后来 Omega（共享状态 + 乐观并发）与 Borg/K8s（集中式 + 抢占）出现的原因。

21.2.9 调度目标与策略：利用率、公平性、延迟与数据局部性

六个调度目标（经常互相冲突）：(1) 高利用率——装箱（bin packing）是其极端形态（first-fit 找第一台装得下的机器、best-fit 找装完剩余最小的机器），但装箱是 NP-hard，且塞得过满会牺牲应对突发与故障转移的弹性；(2) 公平性——常用 Jain 公平指数 $J=\frac{(\sum_i x_i)^2}{n\sum_i x_i^2}\in[1/n,1]$（完全平均为 1，一人独占为 $1/n$，只衡量分布形状、不衡量绝对量）；(3) 低延迟——作业完成时间与排队时间（讲义例：同样任务集 FIFO 平均 14.33 vs STF 9.66，调度策略直接决定延迟）；(4) 数据局部性——把任务放到数据所在机器（本地读 vs 跨机架读可能差一个数量级），为此可能要故意让集群短暂空转，这就是延迟调度（delay scheduling）：队头任务无法在其数据所在节点启动时短暂跳过它，稍后再试（实验表明等约 1 秒即可用极小的公平性代价换取接近最优的局部性）；(5) 容错——发现节点故障并重新调度，要求任务幂等/可重放；(6) 优先级与抢占——高优先级作业可抢占低优先级资源以保证 SLA，代价是重算开销，因此要尽量杀「最年轻的」任务并避免抢占抖动。
常见策略对照表（本章要求的核心表）：

策略	分配规则	是否多资源	公平性	延迟/局部性	抢占	代表系统
FIFO / FCFS	严格按到达顺序，队头拿满再下一个	单资源（或按容器数）	最差：先到者独占，后来者可能饿死	队头阻塞（head-of-line blocking）严重	无	早期 Hadoop 默认队列、单处理器 FIFO
Fair Sharing	让所有作业的（单资源）份额尽量相等；资源空闲时独享，新作业到达后逐步让出	单资源（CPU 槽/容器数）	好	中（可叠加延迟调度改善局部性）	有（可杀任务回收，如 Fair Scheduler 的最小份额）	Hadoop Fair Scheduler、YARN Fair
Capacity Scheduler	队列有保证份额（软限制）与上限（硬限制），队列内通常 FIFO；支持层级队列与弹性借还	单资源为主（可考虑内存）	队列级公平（队内不公平）	中	不允许抢占（只能等任务自然结束）	Hadoop Capacity Scheduler
DRF（主导资源公平）	始终把下一份资源给”主导份额最小”的用户；等化各用户的主导份额	多维（CPU+内存+磁盘+网络……）	理论上最好（满足四条公理）	中（可与延迟/局部性策略叠加）	本身不含抢占（Mesos 里靠 framework 自己或配额实现）	Mesos、YARN 的 DRF 变体、部分云厂商的分布式 OS

DRF 在现实系统里的位置：讲义明确列出——DRF 用于在集群中调度虚拟机（VMs），也用于调度 Hadoop，并且DRF 被 Mesos 采用（”一个面向云环境的操作系统”），同时”类似 DRF 的策略也被一些云计算公司的分布式 OS 使用“。这一句背后的工程事实是：DRF 只解决了”分配多少”（allocation）这一半问题，另一半”放在哪台机器”（placement）由框架调度器或 K8s 的 filter/score 决定——这是 21.2.10 与 21.3.5 结尾要强调的边界。

21.2.10 DRF 与主导份额（Dominant Share）：把一维公平推广到多维

为什么不能只看一种资源：讲义直接用了一个反例式的问题——假设云上的作业有多维需求：Job 1 的每个任务要 2 CPU、8 GB；Job 2 的每个任务要 6 CPU、2 GB。如果只看 CPU 做公平分配（每人一半 CPU = 9 CPU），Job 1 能跑 4 个任务（8 CPU、32 GB）而 Job 2 只能跑 1 个任务（6 CPU、2 GB）：
- Job 1 的 4 个任务要 32 GB 内存，已超出集群总量（例子里是 36 GB，勉强；如果集群是 18 CPU/36 GB 的组合则是 4×8=32 ≤ 36 但已把内存吃光）；
- 而”公平”显然不该让一个用户把另一种资源吃光。
- 反过来只看内存也荒谬：Job 2 的每个任务只要 2 GB，把内存平均分会给它大量任务，从而吃光 CPU。
- 结论：单资源公平在多资源下会产生”一边饿死、一边浪费“的结果。必须有一个同时考虑所有资源维度的公平定义。
主导资源与主导份额的严格定义（本章最核心的两个概念，务必背下）：
- 记集群总容量为向量 $C = (C_1, \dots, C_m)$（例如 $C=(\text{9 CPU}, \text{18 GB})$），用户 $i$ 的每个任务需求为向量 $D_i = (D_{i,1},\dots,D_{i,m})$（例如 $D_A = (1, 4)$ GB）。
- 用户 $i$ 在资源 $r$ 上的需求占比：$D_{i,r}/C_r$——”我的一个任务吃掉这种资源的百分之几”。
- 主导资源（dominant resource）：占比最大的那种资源，即 $\arg\max_r D_{i,r}/C_r$。直观含义：这个用户最”吃”哪种资源（内存密集型、CPU 密集型、IO 密集型）。
- 主导份额（dominant share）：用户当前分配到的各类资源占集群总量的比例中的最大值 $s_i = \max_{r} \frac{x_{i,r}}{C_r}, \qquad x_{i,r} = t_i \cdot D_{i,r}$ 其中 $t_i$ 是用户 $i$ 正在运行的任务数。若任务需求记为 $d_i = \max_r D_{i,r}/C_r$（单任务的主导占比），则 $s_i = t_i \cdot d_i$。用户的”公平份额”由主导份额度量，而不是由任何单一资源度量。
- DRF 的核心规则（一句话）：始终把下一份可用资源分配给当前主导份额最小的用户——这是最大最小公平（max-min fairness）在多资源上的推广：一维 max-min 是”把资源给份额最小的人”，DRF 是”在主导份额这个标量上做 max-min“。
现实类比（讲义之外的助记）：一群朋友分披萨和饮料，每个人的”套餐”比例不同：小 A 每次要 1 块披萨配 4 杯饮料，小 B 每次要 3 块披萨配 1 杯饮料。“平均分披萨”（单资源公平）会让小 B 拿下大量披萨、把饮料全剩给小 A，而小 A 因为”披萨配额”用完而无法开工；DRF 的做法是看每个人”最缺的那样东西”：小 A 最缺饮料、小 B 最缺披萨，于是让”小 A 拿到的饮料比例”与”小 B 拿到的披萨比例”尽量相等。这样既不会有人被饿死，也很少浪费——“不是平均分东西，而是让每个人’瓶颈物资’的满足程度尽量相等”。
机制图解：二维资源平面上的 DRF 分配（用讲义数值例子的解：集群 $C=(\text{9 CPU}, \text{18 GB})$，$D_A=(1,4)$，$D_B=(3,1)$）。

   内存占集群总量比例 (x_B,mem 方向)
    1.0 ┤─────────────────────────────────────────────┬────────┐  C = (9 CPU, 18 GB)
        │                                             │        │
        │        ┌───────────────────┐                │        │
   2/3 ─┤        │                   │  ● A=(3 CPU,12 GB)       │
        │        │   A 的份额方框      │   主导份额 = max(1/3, 2/3) = 2/3
        │        │   （边长 2/3）     │                │        │
        │        │        ┌──────────┼──────┐         │        │
        │        │        │ ● B=(6 CPU, 2 GB)│         │        │
   1/9 ─┤        │        │  主导份额 = max(2/3, 1/9) = 2/3     │
        │        │        │ （B 的份额方框边长也是 2/3）        │
      0 ┼────────┴────────┴──────────────────────────┴────────┴────> CPU 占集群总量比例
        0       1/3      2/3                         1.0
                 ↑                    ↑
            B 的 CPU 占比        A 的 CPU 占比
            6/9 = 2/3            3/9 = 1/3

   关键读法: 每个用户的"主导份额"= 以原点为一角、刚好包住其分配点的【最小正方形】的边长。
             DRF 让两个正方形的边长相等（都是 2/3）⇒ 主导份额相等 ⇒ 多资源公平。
             注意 A 的方框在内存方向"打满"(2/3)，B 的方框在 CPU 方向"打满"(2/3)：
             【每个用户在自己最缺的资源上被满足到同等的程度】。

讲义数值例子（18 CPU / 36 GB）的完整演算：
- 集群 $C = (\text{18 CPU}, \text{36 GB})$；
- Job 1 的每个任务 $D_1 = (\text{2 CPU}, \text{8 GB})$：CPU 占比 $2/18 = 1/9 \approx 0.111$；内存占比 $8/36 = 2/9 \approx 0.222$。因为 $1/9 < 2/9$，所以 Job 1 的主导资源是内存（”Job 1 比 CPU 密集更偏内存密集”），单任务主导占比 $d_1 = 2/9$。
- Job 2 的每个任务 $D_2 = (\text{6 CPU}, \text{2 GB})$：CPU 占比 $6/18 = 1/3 \approx 0.333$；内存占比 $2/36 = 1/18 \approx 0.056$。因为 $6/18 > 1/18$，所以 Job 2 的主导资源是 CPU，单任务主导占比 $d_2 = 1/3$。
- DRF 的性质：讲义原话是”对一个给定作业，它从集群范围内获得的主导资源类型的百分比，对所有作业都相同“，即 $\text{Job 1 的内存占比} = \text{Job 2 的 CPU 占比}$ 这可以写成线性方程并求解。
- 解：设 Job 1 得到 $t_1$ 个任务、Job 2 得到 $t_2$ 个任务，则 $\frac{t_1 \cdot 8}{36} = \frac{t_2 \cdot 6}{18}$，即 $\frac{2t_1}{9} = \frac{t_2}{3}$，也就是 $2t_1 = 3t_2$。可行的最优解是 $t_1 = 3,\ t_2 = 2$：Job 1 得到 3 个任务（6 CPU、24 GB），Job 2 得到 2 个任务（12 CPU、4 GB）；双方的主导份额都是 $2/3$：Job 1 的内存占比 $= 3 \times 8/36 = 24/36 = 2/3$，Job 2 的 CPU 占比 $= 2 \times 6/18 = 12/18 = 2/3$。
- 讲义留下的思考题（很有价值）：”如果云有 120 个 CPU，Job 2 的主导资源是什么？” —— CPU 占比变为 $6/120 = 1/20 = 0.05$，内存占比仍是 $2/36 \approx 0.0556$；由于 $1/20 < 1/18$，主导资源翻转成了内存。这说明：“谁是主导资源”取决于用户需求与集群容量的比值，而不是用户需求的绝对值——把集群扩容成”CPU 相对过剩”，同一个作业就从 CPU 密集型变成了内存密集型，DRF 的分配也随之改变。这是理解 DRF 的关键直觉。
- 一个必须点出的观察（讲义原问题”资源被充分利用了吗？”）：终态用了 18/18 CPU（100%）与 28/36 GB 内存（77.8%，空出 8 GB）。DRF 保证的是帕累托效率，不是”每种资源都打满”：给 Job 1 再加一个任务需要 2 CPU（已无），给 Job 2 再加一个需要 6 CPU（也已无）——没有人能在不损害他人的前提下变好，但仍然有内存闲置。在真实系统中，这些空闲内存会被”别人用不到”这个事实所浪费，因此工程上常把 DRF 与装箱/回填（backfill）或超额分配（overcommit）结合使用。
- 推广性（讲义「其他 DRF 细节」）：DRF 能推广到多个作业与两种以上资源（CPU、RAM、网络、磁盘），并保证每个作业都能得到它最想要的那种资源的公平份额（21.3.5 给出算法与四公理证明，21.4 给出可运行实现）。

21.3 算法伪代码与正确性分析

本节给出五份伪代码，覆盖本章两条主线：流处理的三种容错机制（tuple 级 ack、血缘重算、barrier checkpoint）与集群调度的算法核心 DRF。每份都按”假设与系统模型 → 伪代码 → 逻辑解说（含具体数值走一遍）→ 正确性论证（Safety/Liveness）→ 复杂度”展开。

算法 21.3.1：Storm 的 XOR acking 可靠性机制（Anchoring + Acker + 超时重放）

假设与系统模型

进程模型：crash-stop——bolt/spout 实例可能崩溃并永久丢失其内存中的全部状态（包括”我还没 ack 的 tuple”）；集群中至少一个 acker task 存活（acker 崩溃时该树的校验值丢失，Storm 会把其上的 spout tuple 视为失败并重放）。
通道：tuple 与 ack 消息走可靠通道（TCP），但acked 消息本身可能因为端点崩溃而永远不来；ack 的到达顺序任意（并行分支、网络乱序）。
tuple id：每个 tuple 被赋予一个 64 位随机 id；假设在同一棵树中所有 id 互不相同（碰撞概率见下）。
故障模型：可能的故障是”某个 tuple 永远不被 ack“（节点崩溃、进程被杀）。没有拜占庭故障（节点不会伪造 ack）。
目标：判断”spout tuple 派生的整棵树是否被完整处理”，且每个 spout tuple 的跟踪状态为 $O(1)$。

伪代码

状态（每个 spout root r 在 acker 上维护）：
    V[r] : 64-bit 整数（校验值），初始为 0
    T[r] : 定时器（超时阈值 TIMEOUT）

事件（acker 上）：

upon function CreateTuple(r, tid):              # 某 bolt emit 出一个属于树 r 的新 tuple
    V[r] <- V[r] XOR tid                        # "创建" -> 计入
    if 该 root 是新出现的 then 启动 T[r]

upon function AckTuple(r, tid):                 # 某 bolt/sink 处理完 tuple tid
    V[r] <- V[r] XOR tid                        # "完成" -> 抵消
    if V[r] = 0 then                            # 归零 => 整棵树完成
        取消 T[r]; 通知 spout：ACK(root r); 释放 T[r]

upon function FailTuple(r, tid):                # bolt 显式报错（如数据库异常）
    取消 T[r]; 通知 spout：FAIL(root r); 释放 T[r]     # 立即失败，不等超时

upon event timer T[r] expired:                  # 超时：树中有 tuple 从未被 ack
    通知 spout：FAIL(root r); 释放 T[r]

Spout 侧：

upon 收到 FAIL(root r):
    重新 emit 该 root（重放整棵树，可携带原 tuple 值）

Bolt 侧（OutputCollector）：
    emit(anchor_list, new_values):              # 锚定：新 tuple 属于 anchor 中所有 tuple 所在的树
        为每个 anchor 所在的 root r 发送 CreateTuple(r, tid_new)
        （同时记录 父->子 边，用于失败传播）
    ack(tuple):   对每个所属 root r 发送 AckTuple(r, tuple.tid)
    fail(tuple):  对每个所属 root r 发送 FailTuple(r, tuple.tid)
    ！每个 tuple 必须恰好 ack 或 fail 一次，否则内存泄漏 / 永远超时

算法逻辑解说（用 21.2.4 的树逐步走一遍）

树的结构：$T_0$（spout）派生出 $T_1, T_2$（bolt1）；$T_1$ 派生出 $T_3$、$T_2$ 派生出 $T_4$（bolt2）；$T_3, T_4$ 是叶子（sink）。
事件序列与校验值（id 截断为 16 位便于阅读）：create T0 ^A001 → A001；create T1 ^B002 → 1003；create T2 ^C003 → D000；ack T0 ^A001 → 7001；create T3 ^D004 → A005；ack T1 ^B002 → 1007；create T4 ^E005 → F002；ack T2 ^C003 → 3001；ack T3 ^D004 → E005；ack T4 ^E005 → 0000。
关键时刻：当最后一个 ack 到达时 $V = 0$，acker 立刻通知 spout”整棵树完成”。注意整个过程中没有任何一步需要”查找某个 tuple 是否已 ack”——只有一个 8 字节累加器。
失败路径：假设 bolt2 在处理 $T_3$ 时崩溃。此时 create T3 已经计入，但 ack T3 永远不会到达（$T_3$ 的 id 永远留在 $V$ 里），并且 $T_3$ 的所有后代也不会被创建。于是 $V[r] \neq 0$，超时后 spout 重放 $T_0$——重放会重新走一遍整棵树（新 id），已经产生过的输出可能被再产生一次（$T_4$ 分支可能已经成功写出结果），这正是 at-least-once 中”重复”的来源。
锚定到多个输入时：新 tuple 同时属于多棵树，CreateTuple 要发给每一个 root 的 acker；相应地，它失败时要失败所有相关的树（这就是讲义”一个 tuple 的失败会导致一个或多个 tuple 被重放”的机制来源）。

正确性论证

先定义不变量。对每个 root $r$，记 $P_r$ 为树 $r$ 中”已创建但尚未 ack”的 tuple 集合（pending 集合），$\mathrm{id}(t)$ 为 tuple $t$ 的 64 位 id。断言：

\[\textbf{(I)}\quad V[r] \;=\; \bigoplus_{t \in P_r} \mathrm{id}(t) \qquad (\text{空集的异或规定为 } 0)\]

证明（对事件序列作归纳）：初始 $P_r = \emptyset$、$V[r] = 0$，成立。两类事件：

CreateTuple：$P_r \leftarrow P_r \cup \{t\}$，同时 $V[r] \leftarrow V[r] \oplus \mathrm{id}(t)$。由归纳假设，新值 $= \left(\bigoplus_{s\in P_r^{\text{old}}}\mathrm{id}(s)\right)\oplus \mathrm{id}(t) = \bigoplus_{s \in P_r^{\text{new}}}\mathrm{id}(s)$ ✓（这里用到异或的结合律，把新项并入求和）。
AckTuple：$P_r \leftarrow P_r \setminus \{t\}$，$V[r] \leftarrow V[r]\oplus \mathrm{id}(t)$。由 $\mathrm{id}(t)\oplus\mathrm{id}(t)=0$（自逆），新值恰为去掉该项后的异或 ✓。两者都不依赖事件的到达顺序（交换律）✓。

安全性（Safety：不会”漏报完成”）——也就是”$V[r]=0$ 时树真的完成了”：由不变量 (I)，若 $P_r = \emptyset$ 则 $V[r] = 0$。反之，若 $V[r] = 0$ 而 $P_r \neq \emptyset$（假阳性：树没完成但校验值归零），则意味着 $\bigoplus_{t\in P_r}\mathrm{id}(t)=0$ 而 $P_r$ 非空——这在代数上是可能的（例如 $\mathrm{id}$ 为 1、2、3 的三个 tuple 异或为 0）。但 tuple id 是均匀随机的 64 位数：在给定其余 $k-1$ 个 id 的条件下，最后一个 id 必须精确等于它们的异或，概率为 $2^{-64}$；对同一棵树在任意时刻取并集（union bound），假阳性概率不超过 $\vert P_r\vert \cdot 2^{-64}$，在 $\vert P_r\vert \le 10^6$ 时约 $10^{-13}$。结论：XOR acking 的”归零 ⟺ 完成”是在 $1-2^{-64}$ 量级置信度上的等价，而不是无条件等价——这是本算法唯一一处工程性近似，考试若问”XOR 是否严格等价”，答案必须包含这个限定。安全性还有一条更硬的前提：每个 tuple 必须被 ack 恰好一次。若某个 tuple 被 ack 两次，它的 id 被抵消了两次（净效果是”永远处于 pending”），于是 $V[r]$ 通常不会归零 ⇒ 该树超时并被判失败、整体重放（表现为多余的重复处理）；只有在极小概率下（剩余 pending 集合的异或恰好等于该 id，或 id 碰撞）才会提前归零 ⇒ acker 提前宣布成功 ⇒ 静默丢数据。因此”重复 ack”是一条会同时产生假失败（常见）与假成功（罕见但致命）的 bug 路径，而”漏 ack”只会产生假失败（超时重放）——这正是讲义强调”必须记得对每个 tuple 调用 ack/fail”以及”每个 tuple 都占用内存、忘记确认会导致内存泄漏”的深层原因。

活性（Liveness：不会”永远悬着”）：

若系统无故障且所有 bolt 都正确 ack，则每个 tuple 都被 create 一次、ack 一次，$P_r$ 最终为空，由不变量 $V[r] = 0$，acker 必然在有限时间内通知完成（因为 Create/Ack 事件数是有限的——一棵树的 tuple 数是有限的）✓。
若某个 tuple 永远不会被 ack（节点崩溃），则 $V[r]$ 永远不为 0——校验值本身不会自愈。活性由超时机制提供：定时器 $T[r]$ 到期 ⇒ FAIL ⇒ spout 重放。因此超时阈值的选择是活性与重复量的权衡：阈值太小会把慢任务误判为失败（制造大量重复），阈值太大则故障恢复变慢。讲义对失败的定义正是”在指定超时时间内没有完整处理”。

复杂度

空间：每个 spout root $O(1)$（一个 64 位整数 + 一个定时器）；消息：每个 tuple 产生 2 条记账消息（create + ack），即 $O(\vert T\vert )$，与数据量成正比，是 Storm 吞吐的主要瓶颈；时间：每条记账消息 $O(1)$（一次异或）。
检错延迟：最坏为 TIMEOUT，故故障路径的端到端延迟 = TIMEOUT + 重放路径长度；容错能力：任意节点崩溃（只要 acker 与 spout 存活），因为「重放」不依赖任何已丢失的状态。

算法 21.3.2：Spark 的血缘记录、分区重算与 stage 划分

假设与系统模型

数据模型：数据集是不可变的记录集合，被切成 $P$ 个分区，分布在集群的 worker 上；源数据存放在可靠且可重放的存储（HDFS、可靠接收的流输入）或一个检查点中。
确定性假设：每个转换（transformation）是一个确定性函数 $f$：给定父分区的全部记录，输出确定的分区内容。
故障模型：worker crash-stop（内存中的数据丢失）；不丢源数据（源在 HDFS/多副本，或输入在被确认前已复制到两个 worker）。
目标：故障时只重算丢失的分区，稳态零额外成本（不复制、不落盘）。

伪代码

状态：
    RDD 元数据: R.id, R.partitions[1..P], R.deps = { (父RDD, 分区映射, 转换函数) }
    R.cached : 该 RDD 是否常驻内存
    lineage(r, p) : 分区 (r,p) 的依赖链（由 R.deps 递归得到）

记录阶段（构建 DAG 时，惰性求值）：

upon transformation  child <- f(parent_list, partitioner):
    child.deps <- { (parent_k, partitioner_k, f) } for each parent_k   # 只记元数据，不计算
    return child

执行阶段（遇到 action，生成 job）：

function execute(rdd, action):
    stages <- split_stages(rdd)                    # 见下方 stage 划分
    for st in stages (拓扑序):
        for p in st.partitions:
            schedule_task(st, p)                   # 任务 = 对一个分区施加 st 内的窄依赖链

function split_stages(rdd):
    stages <- [];  cur <- 新建空 stage
    for r in 反向遍历(rdd) 按 R.deps:              # 从 action 向源回溯
        if r 的所有依赖都是窄依赖 (父分区 -> 至多一个子分区):
            cur.加入(r)                            # 可以流水线执行，不出 stage
        else (存在宽依赖 / shuffle):
            提交 cur 为一个 stage; cur <- 新建 stage（以 shuffle 为边界）
    return stages（拓扑序 = 反向回溯后取逆序）

恢复阶段（某个 worker 失败，丢失了它持有的分区集合 LOST）：

upon worker failure:
    LOST <- 该 worker 上所有 cached RDD 分区 + 正在运行的任务的产物
    for (r, p) in LOST:
        if (r,p) 可按 lineage 重算:
            提交重算任务：按 lineage(r,p) 从最近的【可用且可靠】的祖先（源或检查点）开始
            重新施加转换函数链，得到新的 (r,p)
            # 多个 (r,p) 可以并行提交 —— 这就是 D-Streams 的【并行恢复】
        else:
            从检查点读取

血缘截断（lineage cutoff）：
    周期性地对高价值/长链的 RDD 做检查点（例如每 5 个 RDD 复制一次），
    检查点成功后【忘掉该 RDD 的重算血缘】，使其状态不随运行时间无界增长。

算法逻辑解说

stage 划分的直觉：从 action 反向看依赖图。只要一路都是窄依赖，就”不需要打断”——因为父分区算完可以立刻喂给子分区，同一个 thread 里顺着函数链跑下去（流水线）。一旦遇到宽依赖（shuffle），就必须”打断并等”：必须等所有父分区算完、shuffle 数据落好，才能开始下一段的子分区。因此 stage 数 = 从 action 到源路径上宽依赖的个数 + 1。
举例：srcRDD --flatMap--> words --map--> pairs ══reduceByKey══> counts --map--> out。反向遍历：out --map--> counts 是窄依赖；counts 依赖 pairs 是宽依赖（shuffle）⇒ 切一刀，形成 Stage 2 = {counts → out 的窄依赖链} 与 Stage 1 = {src → words → pairs 的窄依赖链}。Stage 1 内部三个转换在一个 task 内流水线执行（不物化中间 RDD）。
恢复举例（对照讲义”没有失败就没有代价”）：假设 worker W2 崩溃，丢失了上一轮 cache 的 pairs 的 1 个分区与 counts 的部分分区。
- 窄依赖部分：pairs[p] 只需从 srcRDD 的对应分区重新跑 flatMap → map，只读该分区的源数据（在 21.4 的代码实验里，源分区 1 只有 4 条记录）。
- 宽依赖部分（有状态算子）：counts 是一个 reduceByKey 的状态 RDD，其重算需要 counts 的上一轮状态。如果上一轮状态还在其它节点内存里，重算只需当前 interval 的数据；如果丢了，就要从最近的检查点重算到当前 interval——D-Streams 论文的 WordCount 实验正是这个情形：因为 reduceByKey 每轮都要”减去 30 秒前的数据”，血缘图会无限增长，所以必须给它配一个检查点间隔（论文用的是 10 秒）。
- 并行恢复的量化好处（论文模型）：设故障前单节点负载为 $\lambda$（以节点算力为单位），从检查点算起有 1 分钟工作需重建。上游备份（单点串行）：$t_{up}\cdot1=\lambda+t_{up}\lambda \Rightarrow t_{up}=\frac{\lambda}{1-\lambda}$；并行恢复（$N$ 节点分担）：$t_{par}\cdot1=\frac{\lambda}{N}+t_{par}\frac{N}{N-1}\lambda \Rightarrow t_{par}\approx\frac{\lambda}{N(1-\lambda)}$。对照：$\lambda=0.8$ 时 $t_{up}=4$，而 $N=5/10/20$ 时 $t_{par}=0.80/0.40/0.20$——并行恢复比上游备份快一个数量级，且不需要 2 倍硬件，这就是「确定性批计算」的红利。

正确性论证

定理（血缘恢复的正确性）：设分区 $(r,p)$ 的血缘链为确定性函数复合 $F = f_k \circ \cdots \circ f_1$，其输入为可靠可重放的源分区集合 $S$。若 $(r,p)$ 丢失，则重算得到的分区内容与原内容逐记录相同。

证明（对血缘链长度归纳）：

基例：$r$ 是源 RDD（长度为 0）。丢失的源分区可从可靠存储（HDFS 多副本 / 已复制到两个 worker 的输入 / 检查点）重新读取，内容相同——这依赖”源可靠”这一假设 ✓。
归纳步：设 $r$ 的分区 $p$ 由父分区集合 $\{(r_1,p_1),\dots,(r_m,p_m)\}$ 经确定性函数 $f$ 生成，即 $x_p = f(y_{p_1},\dots,y_{p_m})$。对每个父分区：若它仍可用，则 $y_{p_i}$ 与原值相同；若它也已丢失，则由归纳假设，重算得到相同的 $y_{p_i}$。由于 $f$ 是确定性函数（相同输入 ⇒ 相同输出），重算出的 $x_p = f(\cdot)$ 与原值逐记录相同 ✓。∎
推论 1（推测执行的安全性）：同一任务的多个副本产生相同输出，因此”谁先完成用谁的结果”不会破坏正确性。
推论 2（并行恢复的安全性）：并行重算的是互不相同的分区，彼此通过血缘解耦（窄依赖时）或通过重放同一 shuffle 输入解耦（宽依赖时），故并行不改变结果。

关键的假设审查（哪些情况会破坏正确性）：

非确定性转换：random()、currentTimeMillis()、未排序的 hash 遍历顺序、读取外部可变状态（如未纳入血缘的数据库查询）都会让 $f(y) \ne x$。工程修正是把非确定性提升为血缘中的显式节点（随机种子、时间戳作为输入 RDD）。
源不可重放：如果输入在被可靠存储之前就被”消费掉”（例如直接从 socket 读且不落盘），则重算无米下锅。D-Streams 论文的对策是输入复制到两个 worker 后才向客户端确认。
血缘记录不全：分区映射信息丢失（例如自定义算子没有正确声明依赖）会导致重算用了错的父分区——Spark 要求自定义 RDD 必须实现 getDependencies 与 compute 两个方法。
血缘无限增长：正确性不受影响，但恢复时间无界（每次都要从很久以前重算），因此必须 checkpoint + lineage cutoff。

复杂度

空间：血缘元数据 $O(V+E)$，与数据量无关；stage 划分 $O(V+E)$（一次反向遍历）。
恢复时间：单分区血缘链长为 $k$ 时重算成本 $\approx\sum_{j=1}^{k}c_j$；丢失 $L$ 个分区且能并行重算、集群有 $N$ 台机器时，墙钟时间约 $\frac1N\sum_{p\in L}\text{cost}(p)$ 加调度开销。
稳态开销：0 次额外 IO、0 次复制（除非显式 persist/checkpoint）；检查点成本 $O(\text{状态大小})$，频率越高恢复越快、稳态开销越大（21.4 给出对照表）。

算法 21.3.3：Spark Streaming 的微批执行与窗口聚合（含 exactly-once 的三个条件）

假设与系统模型

时间：节点时钟经 NTP 同步（偏差毫秒级）；批间隔 $\Delta$ 固定（如 1 s）。输入按”到达时间”归入批（D-Streams 原口径），迟到数据由”松弛时间”或”应用层修正”处理。
输入：可重放（Kafka offset、HDFS 目录、或”接收时复制到两个 worker”的可靠队列）。
故障模型：任意 worker crash-stop；master 通过血缘元数据驱动恢复（论文的实现中 master 本身不做容错，但修复方式是”把 DStream 图与检查点可靠存储 + 备份接管”）。
目标：端到端 exactly-once（在 21.3.3 末尾给出三条件）。

伪代码

状态：
    graph : DStream 血缘图（含窗口与状态算子的定义）
    stateRDDs : 有状态算子在每个 interval 的状态 RDD 序列
    ckpt_interval : 检查点间隔（多少个 interval 做一次）
    offset[batch] : 每个 interval 的输入偏移（用于重放）

主循环（master 每个 interval 触发一次）：

upon interval t 到来（时钟越过 t*Δ）:
    # ---- ① 固定输入：把 [t, t+1) 内到达的记录收集为该 interval 的输入数据集 ----
    recv <- 所有 receiver 上报的本 interval 收到的数据块（已在可靠性层复制）
    inputRDD[t] <- 由这些数据块构成的不可变、可分区 RDD
    offset[t] <- 各输入源的当前偏移

    # ---- ② 由血缘图计算本 interval 的各 DStream 输出 ----
    for each DStream d in graph（拓扑序）:
        按 d.operator 生成本 interval 的 RDD：
          无状态算子:   outRDD[t] <- f(父DStream 的同 interval RDD)
          窗口算子:      outRDD[t] <- f(父DStream 的 RDD[t-L+1 .. t])          # L = windowLength/Δ
          增量窗口:      outRDD[t] <- 合并( 窗口内各 interval 的中间结果 )
          可逆增量窗口:  outRDD[t] <- outRDD[t-1] ⊕ 新进入的 interval ⊖ 滑出的 interval
          有状态算子:    stateRDD[t] <- update(stateRDD[t-1], inputRDD[t])
                         （血缘中显式记录对 stateRDD[t-1] 的依赖）
    # ---- ③ 提交（原子的）----
    在【同一个事务/原子提交】中完成：
        (a) 把输出写往外部系统（幂等写 或 事务性预提交）
        (b) 记录 offset[t]（本 interval 的输入进度）
    二者要么都生效，要么都不生效

    # ---- ④ 周期性检查点 ----
    if t mod ckpt_interval == 0:
        异步地把 graph 元数据 + stateRDD[t] 复制到可靠存储（检查点）
        成功后对已检查点的 RDD 执行 lineage cutoff（忘掉其重算血缘）

恢复流程（worker 失败）：
    LOST <- 失败节点上的 RDD 分区（含 stateRDD 分区）
    并行地从最近的可靠点（源/检查点）重算 LOST 中的每个分区     # 算法 21.3.2 的恢复过程
    对 stateRDD：若失败的 interval 的输出【尚未原子提交】，则整个 interval 的输入重新处理；
                 若已提交，则不需重算（提交是原子的，不会有"半提交"状态）

算法逻辑解说（以”3 秒窗口、1 秒滑动”为例走一遍）

$t=5$ 到来时，window(3s,1s) 的输出 RDD 覆盖 RDD[3..5]；$t=6$ 时覆盖 RDD[4..6]——两个相邻窗口共享 2 个 interval。
朴素实现：每轮对 3 个 interval 重新聚合，$t=6$ 时算 3 次、$t=7$ 时算 3 次……同一 interval 被反复求和。
可逆增量实现（论文的 reduceByWindow(f, g)）：RDD[4..6] = RDD[3..5] ⊕ new(RDD6) ⊖ old(RDD3)——每个 interval 只被计入一次，代价是要求聚合函数 $f$ 有逆元 $g$（count/sum 有，max/min 没有）。
有状态算子（updateStateByKey/mapWithState/论文的 track）：状态本身是 RDD，血缘里显式记录”$state[t]$ 依赖 $state[t-1]$ 与本批输入”。这意味着状态的血缘链会随时间无限增长，因此必须周期性 checkpoint 并做 lineage cutoff（论文 WordCount 实验检查点间隔 10 秒就是这个原因）。
故障时到底重算什么？ 关键字在”提交“：如果一个 interval 的输出还没有原子提交，那么失败后这个 interval 的输入可以整段重放（因为输入可重放），输出不会出现”写了一半”；如果已经原子提交，则该 interval 的结果是可信的，不需要重做。这就把”重复”从语义里消掉了——重复的输入被处理了，但输出的效果只提交了一次。

正确性论证：恰好一次（exactly-once）的三个条件

定理（端到端 exactly-once 的充分条件）：若同时满足

输入可重放（replayable input）：每个 interval 的输入偏移被持久记录，且源系统允许按偏移重读；
状态/血缘可回滚（deterministic recomputation）：所有转换确定性，且丢失的分区可由血缘重算（算法 21.3.2）；
输出幂等或事务性（idempotent or transactional output）：对同一 interval 的输出重复提交不会产生额外效果（幂等键），或输出与外部的偏移记录在同一个事务里原子提交（事务性输出）；那么每个输入记录对最终输出的影响恰好一次。

证明（分情形）：

情形 A：故障发生在输出提交之前。此时外部系统没有观察到该 interval 的任何效果（由条件 3 的原子性/幂等性：未提交的输出被丢弃或可被覆盖）。恢复流程从输入偏移重放该 interval，确定性重算（条件 2）得到与原来相同的输出，然后提交一次。⇒ 效果恰好一次 ✓。
情形 B：故障发生在输出提交之后。提交点（含偏移记录）已经持久化（条件 1、3），因此该 interval 不会再被重放；后续 interval 从已记录的偏移继续。⇒ 该 interval 的效果也恰好一次 ✓。
不存在”提交了一半”的第三种情形：由条件 3 的原子性（事务性输出或幂等写）排除。
归纳：对 interval 逐个应用 A/B，端到端每条记录的效果恰好一次 ✓。∎

反例（说明三个条件缺一不可）：

缺条件 1（输入不可重放，例如”收到就丢”的 UDP 输入）：故障后无法重建丢失的数据，只能丢数据（退化为 at-most-once）。
缺条件 2（转换非确定性，例如输出里带 now()）：重算产生不同结果 ⇒ 外部系统看到”同一条记录两个不同结果”。
缺条件 3（输出是”先写数据库再记录偏移”，且写不幂等）：故障恰好发生在两者之间 ⇒ 恢复后重复插入，产生脏数据。这是工业界最常见的 exactly-once 破功原因。

活性与延迟：只要集群能在 $\Delta$ 内处理完一个 interval 的数据，系统就能持续产出（否则会积压，形成正反馈直到背压/丢弃）。端到端延迟的下界是 $\Delta$（批间隔）加上本批处理时间：即使数据只来一条，也必须等到本批结束——这是微批模型的结构性下限，无法通过工程优化消除。

复杂度

每 interval 的调度开销与「算子数 × 分区数」成正比，批间隔越小固定开销占比越高（这就是批间隔不能无限小的定量原因）；空间：状态 $O(\text{key 数}\times\text{状态大小})$，检查点另需等量的可靠存储。
恢复时间 $\approx$ 需重算的 interval 数 × 单 interval 处理时间 ÷ 并行度，也可用论文模型 $t_{par}\approx\lambda/(N(1-\lambda))$；容错能力：任意 worker（分区可重算），master 本身需额外容错设计。

算法 21.3.4：Flink 式的 Chandy-Lamport barrier checkpoint（= Lecture 12 的 marker）

假设与系统模型

通道：FIFO、可靠（TCP）——这是 Chandy-Lamport 算法的前提：barrier 与普通记录在同一通道上按序流动，因此”barrier 之前到达的记录必然属于快照前状态”。
算子：数据流图是有向图（含分叉与汇合）；算子既可无状态，也可有状态（keyed state/operator state）。
故障模型：crash-stop（算子状态与上游在途数据全部丢失）；JobManager 负责协调检查点（其自身可用高可用元数据存储）。
目标：周期性地获得全局一致的快照（一致割），配合可重放输入与事务性输出实现 exactly-once。

伪代码

源算子 source：
    n <- 0                                     # checkpoint id
    周期性触发（由 JobManager 发起第 n 次检查点）：
        n <- n + 1
        把当前输入偏移记入【快照 n 的元数据】（相当于 Chandy-Lamport 的"记录进程状态"）
        在【每条】输出通道上注入一条 barrier(n)（= Lecture 12 的 marker，携带 id n）

算子 op（有 m 条输入通道 in_1..in_m，若干输出通道）：
    状态：aligned_set <- {}（本轮已到达 barrier 的通道集合）
          buffer[i] <- 通道 i 上 barrier 之后到达的、暂不处理的记录

upon 从通道 i 收到一条普通记录 x:
    if i ∈ aligned_set then
        buffer[i].append(x)                    # 该通道已经对齐，barrier 之后的记录先缓存
    else
        按 op 的语义处理 x（更新状态、向下游发射结果）

upon 从通道 i 收到 barrier(n):
    aligned_set <- aligned_set ∪ {i}
    if aligned_set = {1..m} then               # 【对齐完成】：所有通道的 marker 都到了
        # (1) 快照本地状态（= Chandy-Lamport 的"记录本地状态"）
        把 op 的状态写入检查点 n 的持久化存储
        # (2) 向下游【广播】barrier(n)（= 沿着输出通道继续传播 marker）
        for each out_channel: 发送 barrier(n)
        # (3) 对齐结束，恢复处理被缓存的记录（它们属于快照之后的数据）
        for i in 1..m: 处理 buffer[i] 中的全部记录; buffer[i] <- {}
        aligned_set <- {}
        # (4) 向 JobManager 确认"算子 op 已完成检查点 n"

JobManager：
    upon 收到所有算子对检查点 n 的确认:
        把检查点 n 标记为【已完成】（只有已完成的检查点可用于恢复）
        ！未完成的检查点（例如对齐期间发生故障）直接丢弃

sink（事务性输出，2PC）：
    预提交：把本 epoch 的输出写入临时位置 / 开启事务
    upon 收到本次检查点已完成的通知：真正提交
    upon 故障恢复：丢弃所有未提交的输出

恢复流程：
    回滚：从最近一个【已完成】的检查点 n 恢复所有算子的状态
    重置：把所有源算子的读取位置重置为 barrier(n) 所标记的偏移
    重放：从该偏移继续处理（下游算子状态已是快照时刻的状态）

算法逻辑解说（把 Lecture 12 的对应关系钉死）

barrier 就是 Chandy-Lamport 的 marker：Lecture 12 中，发起者在自己每条输出通道上发送 marker，其它进程在收到第一条 marker 时记录本地状态，并沿自己的输出通道转发 marker；当所有输入通道的 marker 都到齐，该进程的本地状态就是”一致割”的一部分。Flink 的做法一一对应：
- “发起者在每条输出通道上发 marker” ↔ source 在每条输出通道注入 barrier；
- “收到 marker 时记录本地状态” ↔ 对齐完成后 snapshot 算子状态（Flink 选择”收齐所有通道的 marker 再记录”，即对齐，这比 Chandy-Lamport 原始版本的”收到第一条 marker 就记录”更严格，但语义等价，且更适合有确定性状态快照需求的引擎）；
- “沿输出通道转发 marker” ↔ 广播 barrier 到下游；
- “所有进程记录完毕 = 一致割” ↔ 所有算子确认 + JobManager 标记检查点完成。
两处必要的工程扩展（否则无法落地）：
1. 携带 checkpoint id 的多代检查点：barrier(n) 里的 $n$ 让引擎能同时处理多个未完成的检查点，也能识别”这是上一代的 barrier”（丢弃过期检查点），从而不需要停止数据流。
2. 与外部世界的原子性：Chandy-Lamport 只保证”分布式状态的快照一致”，管不住”已经写进数据库的输出”。Flink 用 sink 的 2PC（预提交 + 检查点完成后再提交）把外部输出也纳入快照的原子性范围——这一步才是端到端 exactly-once 的最后一环。
对齐的代价（用代码实验的数字）：两条输入通道分别在 step 4 与 step 6 收到 barrier；通道 A 在 step 5、6 到达的 2 条记录必须缓存等待（不能提前处理，否则会污染快照前状态），等对齐完成后才处理。背压越严重，先到的通道要等的越久，缓存越多，延迟尖峰越大——这就是 aligned checkpoint 的”停顿（stall）”问题。非对齐 checkpoint（unaligned）的修正是：让 barrier 越过队列中已有的记录（插队），把这些 in-flight 记录作为通道状态写进快照（先落盘，不等对齐）。代价是快照更大（包含在途数据），收益是几乎无停顿。这个权衡在 21.5 会再展开。

正确性论证

安全性（快照是一致割）：称快照 $n$ 覆盖的”割”为：每个算子取其在”接收完全部输入通道的 barrier(n) 之后、处理任何 barrier 后记录之前”的状态。

论证：设算子 $A$ 的状态在快照中为 $S_A$，算子 $B$ 的状态为 $S_B$，且 $B$ 的输入之一是 $A$ 的输出通道（FIFO）。需要证明：不存在”$B$ 已经处理了 $A$ 在快照前发出的记录 $x$，但 $A$ 的快照不包含 $x$ 的效果” 这种跨算子不一致。
- 若 $x$ 在 $A$ 的快照之前被 $A$ 发出，则 $x$ 在通道上位于 barrier(n) 之前（FIFO ⇒ barrier 之后不会混入更早发出的记录）。因此 $B$ 在处理自己的”A 通道的 barrier”之前必然已经收到并处理了 $x$（同样是 FIFO：先到 $x$ 后到 barrier）。由于 $B$ 只有在 A 通道的 barrier 到达之后、且所有通道对齐之后才快照状态，$x$ 的效果已经包含在 $S_B$ 中 ✓。
- 反之，若 $x$ 在 $A$ 的快照之后发出，则它位于 barrier 之后，会被 A 通道的 barrier 挡住（进入 buffer 或被视为新 epoch），不会进入 $S_B$ ✓。
- 对多条输入通道、多级算子归纳即得：所有算子的快照 + 各通道的在途记录（对齐式下通过”停读”隐式处理）共同构成一个全局一致的状态切面 ✓。∎ 这条论证唯一依赖的假设是通道 FIFO——这正是 Lecture 12 的原假设，也是为什么 Flink 必须用 TCP 或在应用层保证按序。

活性（有限时间内完成检查点）：

在无故障、无永久背压的条件下，每条通道上的 barrier 都会在有限时间内到达接收算子（通道可靠 + FIFO），因此对齐必然完成，快照必然产生 ✓。
背压下的活性风险：若某条通道被背压”堵死”（下游不消费，barrier 排在队列后面永远到不了），对齐会无限期等待，检查点迟迟无法完成 ⇒ 一旦此时发生故障，只能回滚到更老的检查点，恢复时间与重复量都变大。这正是对齐式 checkpoint 在持续背压下恶化的机理，也是 unaligned checkpoint 存在的理由（barrier 插队 ⇒ 不必等队列消化）。

复杂度

每轮检查点的消息数：每条数据通道 1 条 barrier ⇒ $O(E)$；快照大小：对齐式 $O(\sum\text{算子状态})$，非对齐式还要加 $O(\text{在途数据})$。
对齐停顿：对齐式最坏为「最慢通道 barrier 到达时间 − 最快通道 barrier 到达时间」；非对齐式为 0（不对齐等待），代价转移到检查点大小与 IO 带宽。
容错能力：任意算子崩溃，但受「最近一个已完成的检查点」限制；对齐期间发生故障时该检查点被整体丢弃。

算法 21.3.5：DRF（Dominant Resource Fairness）完整调度算法与四公理证明

这是本章最重要的算法。

假设与系统模型

资源模型：$m$ 种可无限细分（fluid / 可任意分割）的资源，集群总容量向量 $C = (C_1,\dots,C_m)$（例如 CPU、内存、磁盘网络带宽）。这个”可细分”假设是 DRF 全部理论性质的基石，真实系统的”最小分配粒度（例如 K8s 的 100m CPU、YARN 的容器大小）”会造成偏差（见本算法的”局限”一节）。
用户模型：$n$ 个用户（框架/作业/租户），用户 $i$ 提交单一需求向量 $D_i$（其所有任务同构，每个任务要 $D_i$），效用 = 能运行的任务数。用户 $i$ 有可选权重 $w_i$（加权 DRF，用于表达不同的资源配额）。
调度模型：一个中央分配者（Mesos master / Borg 的分配器）可以做全局决策；分配是静态一次性的（在给定需求下求一个可行分配），动态系统则反复求解（每来一个 offer 轮次重算一次）。
目标：一个”公平”的多资源分配——公平的定义由后面四条公理给出。

伪代码

算法 DRF-Fluid (C, {D_i}, {w_i})            # 流体模型：解析解，用于分析
for i in 1..n:                            # 每个用户单任务的主导占比
    d_i <- max_{r=1..m} ( D_i[r] / C[r] )
sigma <- min_{r=1..m} ( C[r] / ( sum_{i=1..n} w_i * D_i[r] / d_i ) )
        # sigma = 所有用户被等化到的【加权主导份额】的公共值
for i in 1..n:
    t_i <- sigma * w_i / d_i              # 用户 i 应得的（可为小数的）任务数
    for r in 1..m:  x_i[r] <- t_i * D_i[r]      # 物化分配向量
return {x_i}, sigma

算法 DRF-Greedy (C, {D_i}, {w_i})           # 离散贪心：Mesos 实际使用的形式
for i in 1..n:  t_i <- 0; x_i <- 0_{m}
loop:
    for r in 1..m:  free[r] <- C[r] - sum_{i} x_i[r]
    E <- { i : 对所有 r, D_i[r] <= free[r] }      # 未阻塞（还能再装一个任务）的用户
    if E = {} then break                            # 所有用户都被阻塞 -> 终止
    i* <- argmin_{i in E} ( ( max_{r} x_i[r]/C[r] ) / w_i )   # 【主导份额最小的用户】
        # 平手时用确定的规则打破（例如用户编号 / 先到先服务）
    t_{i*} <- t_{i*} + 1
    for r in 1..m:  x_{i*}[r] <- x_{i*}[r] + D_{i*}[r]        # 把这一份资源给它
return {x_i}, {t_i}

实现注记（Mesos）：master 维护每个框架当前的分配向量 $x_i$；每轮 offer 用第 6 步选出「主导份额最小」的框架，把当前空闲资源作为 offer 发出去；接受则 $x_i$ 增加并启动任务，拒绝/超时则资源收回、稍后按同一规则 offer 给别人；加权版本只需把第 6 步的比较量除以 $w_i$（等价于给权重大的用户更高的目标份额）。

算法逻辑解说（用 $C=(\text{9 CPU}, \text{18 GB})$、$D_A=(\text{1 CPU}, \text{4 GB})$、$D_B=(\text{3 CPU}, \text{1 GB})$ 逐步走一遍）

先算单任务主导占比： $d_A = \max\!\left(\frac{1}{9},\ \frac{4}{18}\right) = \max(0.111,\ 0.222) = \frac29 \quad(\text{A 的主导资源是内存})$ $d_B = \max\!\left(\frac{3}{9},\ \frac{1}{18}\right) = \max(0.333,\ 0.056) = \frac13 \quad(\text{B 的主导资源是 CPU})$

流体解：$\sigma = \min\left(\dfrac{9}{1/(2/9) + 3/(1/3)},\ \dfrac{18}{4/(2/9) + 1/(1/3)}\right) = \min\left(\dfrac{9}{4.5+9},\ \dfrac{18}{18+3}\right) = \min(0.667,\ 0.857) = \dfrac23$。于是 $t_A = \frac{2/3}{2/9} = 3$、$t_B = \frac{2/3}{1/3} = 2$，与下面的离散贪心结果完全一致。

   Mesos 的 resource offer + DRF 分配过程（每一步都打印主导份额）
   ─────────────────────────────────────────────────────────────────────────────────────
   集群 C = <9 CPU, 18 GB>;   A 的任务 D_A=<1,4> (d_A=2/9≈0.222, 内存主导)
                              B 的任务 D_B=<3,1> (d_B=1/3≈0.333, CPU 主导)

   步骤  本轮 offer 给谁        给出后的主导份额                累计占用        剩余可用
         （主导份额最小者）      A          B                   (CPU,GB)       (CPU,GB)
   ────  ──────────────────  ─────────  ─────────           ────────────   ───────────
    1    A  (0.000 vs 0.000)   0.222      0.000               ( 1,  4)       ( 8, 14)
    2    B  (0.222 vs 0.000)   0.222      0.333               ( 4,  5)       ( 5, 13)
    3    A  (0.222 vs 0.333)   0.444      0.333               ( 5,  9)       ( 4,  9)
    4    B  (0.444 vs 0.333)   0.444      0.667               ( 8, 10)       ( 1,  8)
    5    A  (0.444 vs 0.667)   0.667      0.667               ( 9, 14)       ( 0,  4)
    6    终止：A 想再要 1 CPU -> 不足；B 想再要 3 CPU -> 不足（两者都被阻塞）
   ────  ──────────────────  ─────────  ─────────           ────────────   ───────────
   终态: A 得 3 个任务 (3 CPU, 12 GB)，B 得 2 个任务 (6 CPU, 2 GB)，两者主导份额均为 2/3
         利用率: CPU 9/9 = 100%，内存 14/18 = 77.8%（剩余 4 GB，但无人能再用它 —— 帕累托最优）

   可读性提示: 第 1 步两人都是 0，平手；规则需要一个确定的打破方式（此处按用户编号取 A）。
             第 5 步之后两人的份额精确相等（2/3），这是 DRF 的"正常终局"：
             所有未被阻塞的用户主导份额相等；被阻塞者（资源不够再装一个任务）可以低于该值。

正确性论证：四条公理逐一证明

先固定记号：$x_i = t_i D_i$ 为用户 $i$ 的分配；$u_i(x) = \min_r \frac{x_r}{D_{i,r}}$ 表示”分配 $x$ 里能装下多少个用户 $i$ 的任务”（即用户 $i$ 的效用）；DRF 的流体分配为 $x_i = \sigma w_i D_i/d_i$（无权重时 $w_i=1$，$t_i = \sigma/d_i$），且 $u_i(x_i) = t_i = \sigma w_i/d_i$。

公理 1：共享激励（Sharing Incentive）——”每个用户分享集群，不应比独占自己的 $1/n$ 份额更差”。

形式化：设用户 $i$ 独占整个集群能跑 $t_i^{solo} = 1/d_i$ 个任务（因为 $d_i$ 是单任务的主导占比，独占时主导资源先被打满，$t_i^{solo} d_i = 1$）。公理要求 $u_i(x_i) \ge \frac1n t_i^{solo} = \frac{1}{n d_i}$。
证明：只需证 $\sigma \ge 1/n$。对任意资源 $r$，由 $d_j = \max_{r^{\prime}} D_{j,r^{\prime}}/C_{r^{\prime}} \ge D_{j,r}/C_r$ 得 $\frac{D_{j,r}}{d_j} \le C_r$，于是 $\sum_{j=1}^n \frac{D_{j,r}}{d_j} \le \sum_{j=1}^n C_r = n C_r \;\Longrightarrow\; \frac{C_r}{\sum_j D_{j,r}/d_j} \ge \frac1n$ 对所有 $r$ 成立，取 $\min$ 得 $\sigma \ge 1/n$。再由 $u_i(x_i) = \sigma/d_i \ge \frac{1}{n d_i} = \frac1n t_i^{solo}$ ✓。∎
含义：没有任何用户有理由”退出共享、自己拿 1/n 的资源单干”。这条公理是”多租户系统能存在”的经济学基础——它同时否决了”按到达顺序分配（FIFO）”与”按单资源公平”这两类方案（它们都可能让某用户低于 $1/n$ 的独占收益）。

公理 2：策略防伪（Strategy-Proofness / 真实性）——”如实上报需求是最优策略”。

形式化：用户 $i$ 的真实需求为 $D_i$，但它可以上报任意 $D^{\prime}_i$；设它上报 $D^{\prime}_i$ 后拿到分配 $x^{\prime}_i$，则它真正能跑的任务数为 $\hat u_i = \min_{r} \frac{x^{\prime}_{i,r}}{D_{i,r}}$ 公理要求：对所有可能的上报 $D^{\prime}_i$，都有 $\hat u_i \le u_i(x_i) = t_i$（如实上报时的任务数）。注意效用必须用”真实需求”折算——这正是本公理的实质：你拿到的是”你声称要的东西”，而你能用的是”你真正需要的东西”，二者之间的差额就是谎报的代价。
第一步：先给出 DRF 分配的结构性刻画。设上报 $D^{\prime}i$ 时的单任务主导占比为 $d^{\prime}_i = \max_r D^{\prime}{i,r}/C_r$，定义归一化声明向量（claim vector） $e_r \;=\; \frac{D^{\prime}_{i,r}}{d^{\prime}_i}, \qquad \text{由 } d^{\prime}_i \text{ 的定义立得 } \max_r \frac{e_r}{C_r} = 1,\ e_r \le C_r$ 而 DRF 给用户 $i$ 的分配恰好是 $x^{\prime}_i = t^{\prime}_i D^{\prime}_i = \frac{\sigma^{\prime} w_i}{d^{\prime}_i}D^{\prime}_i = \sigma^{\prime} w_i\, e$。即：任何用户在上报任何需求时，其分配都是 $\sigma^{\prime} w_i$ 乘上一个”声明向量”，而声明向量在它自己声明的瓶颈资源上必然打满到 $C_r$。
第二步：导出”谎报收益的上界”。取 $i$ 的真实主导资源 $r_i$（满足 $D_{i,r_i}/C_{r_i} = d_i$），则 $\hat u_i = \min_r \frac{\sigma^{\prime} w_i e_r}{D_{i,r}} \le \frac{\sigma^{\prime} w_i e_{r_i}}{D_{i,r_i}} = \frac{\sigma^{\prime} w_i e_{r_i}}{d_i C_{r_i}} \le \frac{\sigma^{\prime} w_i}{d_i}$ （第一个不等号把 $\min$ 松到 $r_i$ 这一项，第二个不等号用 $e_{r_i} \le C_{r_i}$）。结论：谎报能拿到的真实任务数上界，就是把”真实公式”里的 $\sigma$ 换成”谎报后的全局等份份额” $\sigma^{\prime}$——因此谎报要想有收益，必须让它抬高全体用户的等份份额 $\sigma$，而这需要它的声明向量在”聚合意义”上变得更便宜；可一旦声明变便宜，同一个声明向量就无法再覆盖它的真实需求（$\rho_i<1$），两个效应互相抵消。
第三步：两个可以严格证明的特殊情形：
1. 按比例缩放声明（$D^{\prime}i = \alpha D_i$，$\alpha>0$）**：此时 $d^{\prime}_i = \alpha d_i$，$e_r = \frac{\alpha D{i,r}}{\alpha d_i} = \frac{D_{i,r}}{d_i}$ 与真实情形完全相同，因此 $\sigma^{\prime} = \sigma$，而 $\hat u_i = \min_r \frac{\sigma^{\prime} e_r}{D_{i,r}} = \min_r \frac{\sigma \cdot (D_{i,r}/d_i)}{D_{i,r}} = \frac{\sigma}{d_i} = t_i \quad\text{（精确相等）}$ 即 **“把需求乘上任何倍率”既不赚也不亏（分配 $x^{\prime}_i=\sigma^{\prime} e$ 在真实需求下恰好能跑 $t_i$ 个任务）——这解释了为什么”虚报成大作业”或”把作业拆小”都无利可图（拆分只是把 $\alpha$ 变小而已），也正是公理 1 中”不能通过拆分作业获得更多资源”的来源。
2. 在某一维少报：$\rho_i = \min_r D^{\prime}{i,r}/D{i,r} < 1$，此时 $\hat u_i \le \rho_i \cdot t^{\prime}_i$：拿到手的任务数虽然可能更多，但每个”任务”分到的资源都不够真实需求，折算后反而变少（21.2.10 的直觉：谎报是”用真任务的资源量去跑虚报的任务”）。
第四步：数值验证。一般情形的完整证明较长（原始论文在其模型假设下给出），因此这里补一组大规模穷举检查：对 2 万组随机集群配置（2–4 个用户、随机需求与权重）× 每组 60 种谎报向量（全维多报、全维少报、逐维随机缩放），共约 113 万次”谎报 vs 如实”对照，没有一次谎报能超过如实上报（最大相对收益 0.0）。21.4 的代码 21.4.1 给出了可复现的小规模版本：在 $C=(9,18)$、两用户真实需求均为 $(1,4)$ 的设定下扫描 576 种谎报，如实上报得 $2.25$ 个任务，最好的谎报仍然只有 $2.25$。
诚实声明：本公理依赖 NSDI 2011 模型的假设——任务同构、效用 = 能跑的任务数、用户只能提交单一需求向量、不能把任务拆成需求不同的小任务（拆分已被上述「按比例缩放」引理覆盖）。在这些假设之外（任务间有依赖、可提交异构任务），策略防伪性会削弱——这也是真实系统要在 DRF 之上叠加配额、优先级与抢占的原因。

公理 3：无嫉妒（Envy-Freeness）——”没有用户羡慕别人的分配”。

形式化：对所有 $i \ne j$，$u_i(x_j) \le u_i(x_i)$：用户 $i$ 用自己的需求去衡量用户 $j$ 的分配，得到的任务数不会超过自己那份分配。
证明（这是 DRF 最漂亮的一步，只用到”$d_j$ 是 $j$ 的最大资源占比”）：
- 由 $x_j = t_j D_j$ 与 $t_j = \sigma w_j / d_j$（无权重时 $w=1$）： $u_i(x_j) = \min_r \frac{x_{j,r}}{D_{i,r}} = \min_r \frac{\sigma w_j D_{j,r}}{d_j D_{i,r}} \le \frac{\sigma w_j D_{j,r_i}}{d_j D_{i,r_i}}$ 这里取 $r = r_i$（$i$ 的真实主导资源，满足 $D_{i,r_i}/C_{r_i} = d_i$），把 $\min$ 松成一个具体的项。代入 $D_{i,r_i} = d_i C_{r_i}$ 得 $u_i(x_j) \le \frac{\sigma w_j D_{j,r_i}}{d_j\, d_i C_{r_i}} = \frac{\sigma w_j}{d_i}\cdot\frac{D_{j,r_i}/C_{r_i}}{d_j} \le \frac{\sigma w_j}{d_i} \le \frac{\sigma w_i}{d_i} = u_i(x_i)$ 最后一步用到 $w_j \le w_i$（无权重时 $w_i=w_j=1$，不等式取等号即得结论）∎。这一步合法的原因是：把 $\min_r$ 放松到「$i$ 自己需求占比最大的那一维」只会让上界变大（$\min \le$ 任意一项），因此不等号方向安全。
加权 DRF 的正确表述：当权重不同时，上面的推导给出 $u_i(x_j) \le \frac{w_j}{w_i} u_i(x_i)$——即 加权 DRF 满足”按权重折算后的无嫉妒”：用户 $i$ 不会羡慕”把 $j$ 的分配按 $w_i/w_j$ 缩放后”的那份分配。代码实验验证：加权配置下朴素嫉妒对 = 1（确实存在），而按权重折算后的嫉妒对 = 0（详见 21.4 的实验 3）——这是一个容易被忽略但很重要的细节。
粒度反例（诚实的边界）：上述证明是流体模型下的结果。离散贪心版本在任务粒度较粗时会违反无嫉妒。反例：集群 $C = (10,10)$，用户 small 的任务 $D=(\text{1 CPU},\text{1 GB})$、用户 huge 的任务 $D=(\text{9 CPU},\text{9 GB})$。贪心第 1 步两人份额都是 0，先给 small（1 个任务，份额 0.1），再给 huge（1 个任务，份额 0.9），此后 CPU 与内存都被占满，两人都被阻塞 ⇒ 终态 small = 1、huge = 1。此时 small 羡慕 huge：huge 的分配 (9,9) 里能装下 9 个 small 的任务，而 small 自己只有 1 个（代码实验中该配置报告 1 个嫉妒对）。而流体解给出 small = 5.0、huge ≈ 0.56，无嫉妒对。结论：DRF 的公理在流体模型下成立；离散系统的粒度偏差要靠在 DRF 之上叠加”最小分配单位 + 抢占/再平衡”来缓解。

公理 4：帕累托效率（Pareto Efficiency）——”不能在不损害任何用户的前提下改进某个用户”。

形式化：不存在可行分配 $y$ 使得 $u_i(y) \ge u_i(x_i)$ 对所有 $i$ 成立，且至少一个 $i$ 严格 $>$。
证明（用”DRF 最大化最小效用”这一极值性质）：
1. 先证 DRF 的最大最小性：设 $\sigma^{\prime} > \sigma$，考虑”把所有人等化到份额 $\sigma^{\prime}$”的分配 $t^{\prime}i = \sigma^{\prime} w_i/d_i$。取使 $\sigma$ 取到最小值的资源 $r^\*$（即 $\sigma = \frac{C{r^\}}{\sum_j w_j D_{j,r^\}/d_j}$），则该分配对资源 $r^\*$ 的总需求为 $\sum_i t^{\prime}_i D_{i,r^\*} = \sigma^{\prime} \sum_i \frac{w_i D_{i,r^\*}}{d_i} = \sigma^{\prime} \cdot \frac{C_{r^\*}}{\sigma} = \frac{\sigma^{\prime}}{\sigma} C_{r^\*} > C_{r^\*}$ 即超出容量 ⇒ 不可行。因此不存在把公共的（最小）主导份额提高到 $\sigma$ 以上的可行分配 ⇒ DRF 在”最大化最小（加权）主导份额”意义下最优（max-min 最优）。
2. 由 max-min 最优推出帕累托效率：反设存在 $y$ 帕累托优于 $x$。因为 DRF 的分配让所有用户的效用都等于 $\sigma$（$u_i(x_i) = \sigma w_i/d_i$ 与主导份额成正比；无权重时都等于 $\sigma$），”无人更差”意味着 $\min_i \frac{u_i(y)}{w_i} \ge \min_i \frac{u_i(x_i)}{w_i} = \sigma$，而”某人严格更好”意味着某个 $i$ 有 $\frac{u_i(y)}{w_i} > \sigma$，从而 $\min_i \frac{u_i(y)}{w_i} > \sigma$——这与第 1 步”$\sigma$ 是不可超越的最小效用上界”矛盾 ✓。∎
直觉：任何「让某人更好」的分配都必须从别人那里拿资源，因此不存在无偿改进——这正是帕累托效率。
注意：帕累托效率不意味着每种资源都被用满。讲义数值例子（18 CPU/36 GB ⇒ 用满 CPU、剩 8 GB 内存）与 21.2.10 的 $C=(9,18)$ 例子（剩 4 GB 内存）都是明证：空闲的那部分资源无法在”不损害他人”的前提下被利用（想给 A 再加任务，瓶颈是 CPU 而不是内存）。把”DRF 最优”误解为”利用率最高”是本章最常见的错误。

DRF 的四公理一览表（本章要求）

公理	形式化定义	通俗含义	防住了什么
共享激励	$u_i(x_i) \ge \frac1n u_i(C)$	一起用集群不比「自己拿 $1/n$ 单干」差	FIFO/单资源公平造成的饿死
策略防伪	$\forall D^{\prime}_i:\ \hat u_i(D^{\prime}_i) \le u_i(D_i)$	谎报需求占不到便宜	用户的博弈行为
无嫉妒	$\forall i,j:\ u_i(x_j) \le u_i(x_i)$（加权时按 $w_i/w_j$ 折算）	不羡慕别人那份分配	主观不公平感
帕累托效率	不存在可行 $y$：$\forall i\ u_i(y)\ge u_i(x_i)$ 且 $\exists i\ u_i(y)>u_i(x_i)$	没有白白浪费的改进空间	浪费（≠ 每种资源用满）

DRF 的局限（真实系统为什么还要在它之上加东西）

假设资源可无限细分：真实系统有最小分配粒度（K8s 的 100m CPU、YARN 容器、整卡 GPU），粒度粗时 DRF 的性质会退化（10/10 反例）。
不处理放置约束：DRF 只回答「分配多少」，不回答「放在哪台机器」；数据局部性、反亲和、GPU 型号、可用区等由另一层解决（YARN 的 ContainerAllocator、K8s 的 filter+score）。
不考虑任务时长与优先级：效用是「任务数」，隐含假设所有任务等价；真实作业里 10 秒与 10 小时的任务混在一起，长任务会长期占着份额 ⇒ 必须叠加优先级、抢占、配额。
不做动态再平衡：已得的用户不会主动让出，流式到达的作业会让先到者长期占优 ⇒ 工程上用周期性重算目标分配 + 抢占超配额任务修正。
策略防伪的模型假设与算法开销见公理 2 与 21.5。

复杂度

DRF-Fluid：$O(nm)$（计算 $d_i$、$\sigma$、$t_i$）——线性，极便宜。
DRF-Greedy：主循环执行 $\sum_i t_i$ 次（每轮分配一个任务）；每轮需要 $O(nm)$ 计算空闲向量、阻塞集合与最小份额 ⇒ 总时间 $O\!\left(nm \sum_i t_i\right)$。注意 $\sum_i t_i$ 是任务总数，所以离散版本的复杂度与任务数线性相关——这正是 Mesos 用”offer 一批资源（而不是一个一个任务）”来加速的原因：一次 offer 可以启动多个任务，把 $O(\text{任务数})$ 的记账摊薄。
空间：$O(nm)$（每个用户的分配向量），与集群机器数无关（机器级放置由别的组件负责）。
可扩展性边界：DRF 的算法开销可忽略；真正的规模瓶颈是（a）offer 往返的延迟（拒绝/超时会让资源空转）与（b）中央分配器的状态规模（Omega 之所以改成共享状态、K8s 之所以支持调度器分片，都是在解这两个瓶颈）。

21.4 代码示例与分布式实现

本节三个程序都只使用 Python 标准库、random.seed() 固定随机种子、直接 python3 文件名.py 即可运行并打印出可见结果（含断言校验）。它们分别对应本章的两条主线：代码 21.4.1 是 DRF 调度器（最重要的算法实现），代码 21.4.2 是三种流处理容错机制的对照模拟，代码 21.4.3 是集群调度策略（FIFO / Fair / DRF）的仿真对比。

代码 21.4.1：DRF 调度器完整实现（主导份额 + 逐任务分配 + 四公理验证）

#!/usr/bin/env python3
"""DRF (Dominant Resource Fairness) 调度器：主导份额、逐任务分配、四公理验证。
仅用标准库，直接 python3 drf_scheduler.py 运行（随机种子固定，输出可复现）。
资源向量 (cpu, mem)；集群容量 C；用户 u 的每任务需求 D_u；权重 w_u（加权 DRF）。
"""
import random

RES = ("cpu", "mem")


class User:
    def __init__(self, name, demand, weight=1.0):
        self.name, self.demand, self.weight = name, dict(zip(RES, demand)), weight
        self.alloc = {r: 0.0 for r in RES}
        self.tasks = 0.0

    def dominant_ratio(self, cap):                 # d_u = max_r D_ur / C_r
        return max(self.demand[r] / cap[r] for r in RES)

    def dominant_share(self, cap, alloc=None):     # 加权主导份额
        a = self.alloc if alloc is None else alloc
        return max(a[r] / cap[r] for r in RES) / self.weight

    def fits(self, cap, used):
        return all(used[r] + self.demand[r] <= cap[r] + 1e-9 for r in RES)

    def take(self):
        for r in RES:
            self.alloc[r] += self.demand[r]
        self.tasks += 1

    def reset(self):
        self.alloc = {r: 0.0 for r in RES}
        self.tasks = 0.0


def used_of(users):
    return {r: sum(u.alloc[r] for u in users) for r in RES}


def fmt(vec):
    return "(" + ",".join(f"{vec[r]:.3g}" for r in RES) + ")"


def run_policy(users, cap, key, trace=False):
    """通用逐任务循环：每轮把资源给 key 最小且仍装得下的用户，直到无人可装。"""
    for u in users:
        u.reset()
    step = 0
    while True:
        cands = [u for u in users if u.fits(cap, used_of(users))]
        if not cands:
            break
        pick = min(cands, key=key(cap))
        pick.take()
        step += 1
        if trace:
            shares = "  ".join(f"{v.name}={v.dominant_share(cap):.4f}" for v in users)
            print(f"  step {step}: +1 task -> {pick.name:3s} | {shares} | used={fmt(used_of(users))}")
    return step


def drf_policy(users, cap, trace=False):           # 规则：选主导份额最小者
    return run_policy(users, cap, lambda c: (lambda u: u.dominant_share(c)), trace)


def cpu_only_fair(users, cap):                     # 只按 CPU 份额做 max-min
    return run_policy(users, cap, lambda c: (lambda u: u.alloc["cpu"] / c["cpu"] / u.weight))


def fifo_policy(users, cap):                       # 按到达顺序，前一个吃饱再给下一个
    for u in users:
        u.reset()
    for u in users:
        while u.fits(cap, used_of(users)):
            u.take()


def util(users, cap):
    u = used_of(users)
    return {r: u[r] / cap[r] for r in RES}


def jain(values):                                  # Jain 公平指数，0 也参与
    xs = list(values)
    s = sum(xs)
    if s <= 0:
        return 0.0
    return s * s / (len(xs) * sum(v * v for v in xs))


def fluid_drf(users, cap):
    """流体模型：等化加权主导份额, t_u = sigma*w_u/d_u, sigma 取可行上界。"""
    sigma = min(cap[r] / sum(u.weight * u.demand[r] / u.dominant_ratio(cap) for u in users)
                for r in RES)
    return sigma, [{r: sigma * u.weight * u.demand[r] / u.dominant_ratio(cap) for r in RES}
                   for u in users]


def tasks_in(u, alloc):                            # 一份分配能跑多少个"真实"任务
    return min(alloc[r] / u.demand[r] for r in RES)


def report(title, users, cap):
    print(f"  {title}")
    for u in users:
        print(f"    {u.name:7s} tasks={u.tasks:5.1f} alloc={fmt(u.alloc):>16s} "
              f"wdom_share={u.dominant_share(cap):.3f}")
    ut = util(users, cap)
    print("    util: " + " ".join(f"{r}={ut[r]*100:5.1f}%" for r in RES)
          + f" | Jain(tasks)={jain(u.tasks for u in users):.3f}"
          + f" | Jain(wdom)={jain(u.dominant_share(cap) for u in users):.3f}"
          + f" | min_tasks={min(u.tasks for u in users):.1f}")


def envy_pairs(users, cap, allocs, weighted=False):
    """无嫉妒：用户 i 用【自己的需求】去量 j 的分配，不应比自己的更多。
    weighted=True 时按 w_i/w_j 折算 j 的分配（加权 DRF 的无嫉妒形式）。"""
    bad = []
    for i, ui in enumerate(users):
        mine = tasks_in(ui, allocs[i])
        for j, uj in enumerate(users):
            scale = ui.weight / uj.weight if weighted else 1.0
            other = {r: allocs[j][r] * scale for r in RES}
            if tasks_in(ui, other) > mine + 1e-9:
                bad.append((ui.name, uj.name))
    return bad


# ============ 实验 1：讲义数值例子逐步演算 ============
print("=" * 80)
print("实验 1：DRF 逐步分配  集群 <9 CPU, 18 GB>, A:<1,4>, B:<3,1>")
print("=" * 80)
cap1 = {"cpu": 9.0, "mem": 18.0}
us1 = [User("A", (1, 4)), User("B", (3, 1))]
for u in us1:
    dom = max(RES, key=lambda r: u.demand[r] / cap1[r])
    print(f"  {u.name}: D={fmt(u.demand)}  d_u={u.dominant_ratio(cap1):.4f}  主导资源={dom}")
drf_policy(us1, cap1, trace=True)
sigma, al = fluid_drf(us1, cap1)
print(f"  离散终态: A={us1[0].tasks:.0f} tasks, B={us1[1].tasks:.0f} tasks, "
      f"主导份额 {us1[0].dominant_share(cap1):.4f} / {us1[1].dominant_share(cap1):.4f}")
print(f"  流体解:   sigma={sigma:.4f}  A={fmt(al[0])}={tasks_in(us1[0], al[0]):.3f} tasks  "
      f"B={fmt(al[1])}={tasks_in(us1[1], al[1]):.3f} tasks")
print("  利用率:   " + " ".join(f"{r}={v*100:.1f}%" for r, v in util(us1, cap1).items())
      + "   <- CPU 用尽而内存空闲：DRF 保证帕累托效率，不保证用满每种资源")

# ============ 实验 2：DRF vs 单资源公平 vs FIFO ============
print("=" * 80)
print("实验 2：三策略对比  集群 <36 CPU, 36 GB>，U4 权重=2")
print("=" * 80)
cap2 = {"cpu": 36.0, "mem": 36.0}
specs = [("U1_mem", (1, 3), 1.0), ("U2_cpu", (3, 1), 1.0),
         ("U3_bal", (2, 2), 1.0), ("U4_mem2", (1, 2), 2.0)]
for title, fn in (("DRF（加权，选主导份额最小者）", lambda us: drf_policy(us, cap2)),
                  ("单资源公平（只均分 CPU）", lambda us: cpu_only_fair(us, cap2)),
                  ("FIFO（先到先得，队头阻塞）", lambda us: fifo_policy(us, cap2))):
    users = [User(*s) for s in specs]
    fn(users)
    report(title, users, cap2)

# ============ 实验 3：公理验证 ============
print("=" * 80)
print("实验 3：公理验证 —— 无嫉妒 / 最大化最小主导份额 / 策略防伪")
print("=" * 80)
random.seed(42)
fluid_bad = greedy_bad = 0
for _ in range(400):
    cap = {"cpu": random.uniform(8, 30), "mem": random.uniform(8, 30)}
    users = [User(f"u{k}", (random.uniform(1, 8), random.uniform(1, 8)))
             for k in range(random.randint(2, 5))]              # 无权重的纯 DRF
    _, al = fluid_drf(users, cap)
    fluid_bad += len(envy_pairs(users, cap, al))
    drf_policy(users, cap)
    greedy_bad += len(envy_pairs(users, cap, [dict(u.alloc) for u in users]))
print(f"  400 组随机无权重配置: 流体 DRF 嫉妒对={fluid_bad}  离散贪心 DRF 嫉妒对={greedy_bad}")

cap3 = {"cpu": 10.0, "mem": 10.0}
us3 = [User("small", (1, 1)), User("huge", (9, 9))]
drf_policy(us3, cap3)
s3, al3 = fluid_drf(us3, cap3)
print(f"  粒度反例 <10,10>: 离散 -> small={us3[0].tasks:.0f} huge={us3[1].tasks:.0f} "
      f"嫉妒对={envy_pairs(us3, cap3, [dict(u.alloc) for u in us3])}")
print(f"                    流体 -> small={tasks_in(us3[0], al3[0]):.2f} "
      f"huge={tasks_in(us3[1], al3[1]):.2f} 嫉妒对={envy_pairs(us3, cap3, al3)}")

cap4 = {"cpu": 36.0, "mem": 36.0}
us4 = [User(*s) for s in specs]
drf_policy(us4, cap4)
_, al4 = fluid_drf(us4, cap4)
print(f"  加权配置: 朴素嫉妒对={len(envy_pairs(us4, cap4, al4))} "
      f"按权重折算后的嫉妒对={len(envy_pairs(us4, cap4, al4, weighted=True))}  (加权 DRF)")

cap5 = {"cpu": 9.0, "mem": 18.0}
true_users = [User("A", (1, 4)), User("B", (1, 4))]
_, al5 = fluid_drf(true_users, cap5)
truth = tasks_in(true_users[0], al5[0])
best, best_lie = truth, None
for c in range(1, 25):                                   # 扫描 576 种谎报需求
    for m in range(1, 25):
        liar = [User("A", (c / 4, m / 4)), User("B", (1, 4))]
        _, la = fluid_drf(liar, cap5)
        gain = tasks_in(true_users[0], la[0])
        if gain > best + 1e-9:
            best, best_lie = gain, (c / 4, m / 4)
print(f"  诚实上报 A 得 {truth:.4f} task；576 种谎报中最多得 {best:.4f} task（最优谎报={best_lie}）")

sigma5, al5 = fluid_drf(true_users, cap5)
over = [{r: al5[k][r] * 1.01 for r in RES} for k in range(2)]   # sigma 提高 1% 是否仍可行
feasible = all(sum(over[k][r] for k in range(2)) <= cap5[r] + 1e-9 for r in RES)
print(f"  最大化最小主导份额: sigma={sigma5:.4f}；把 sigma 提高 1% 后可行? {feasible}")
assert fluid_bad == 0 and best <= truth + 1e-9 and not feasible
print("\n断言全部通过：流体 DRF 无嫉妒、最大化最小主导份额、谎报无收益。")

【代码做什么？】

定义资源模型：RES = ("cpu", "mem")；User 持有每任务需求 demand、权重 weight、当前分配 alloc 与任务数 tasks；dominant_ratio 计算 $d_u=\max_r D_{u,r}/C_r$，dominant_share 计算（加权）主导份额 $s_u=\max_r x_{u,r}/C_r/w_u$。
通用逐任务分配循环 run_policy(users, cap, key)：每轮算出空闲资源、筛出「还装得下」的用户（未阻塞集合 $E$），把资源给 $key$ 最小的用户，直到 $E$ 为空。DRF 只是把 $key$ 设成 dominant_share——因此同一框架下换个 $key$ 就表达了「单资源公平」（cpu_only_fair）。
实验 1：$C=(9,18)$、$D_A=(1,4)$、$D_B=(3,1)$，打印主导资源、$d_u$、五步分配与每步的主导份额、终态、流体解与资源利用率。
实验 2：$C=(36,36)$、4 个异构用户（含权重 2 者），对比 DRF / 单资源公平 / FIFO，输出任务数、分配、主导份额、CPU 与内存利用率、Jain(tasks)、Jain(加权主导份额)、最小任务数。
实验 3：400 组随机配置检查无嫉妒（用「$i$ 的真实需求能装下 $j$ 的分配多少任务」作效用），分别统计流体解与离散贪心解的嫉妒对；构造粒度反例 $(10,10)$/$(1,1)$/$(9,9)$；统计加权下的朴素嫉妒对与按权重折算后的嫉妒对；扫描 576 种谎报检查策略防伪；把 $\sigma$ 提高 1% 验证不可行（最大最小性 ⇒ 帕累托效率）。

【分布式机制透视】

资源池抽象：代码把集群当成一个可细分的资源向量，run_policy 扮演 Master 侧的分配决策，而「放在哪台机器」被刻意抽象掉——因为 DRF 只解决分配，不解决放置。
消息与状态：这是单进程中央分配器模型（等价于 master 的本地视图）。真实系统每一次「给谁资源」都要经过 resource offer 的一次网络往返，框架还可能拒绝或超时，于是资源存在「空转窗口」——这正是 DRF「算法便宜、往返贵」的原因（21.5）。
并发与一致性：DRF 的状态是每个用户的分配向量；在共享状态调度（Omega）中它会被多个调度器乐观并发修改（冲突重试），本代码则是一次串行循环（等价于单体调度器）。
对真实系统的映射：User ↔ Mesos framework / YARN queue / K8s namespace+quota；demand ↔ 容器资源请求（K8s requests）；weight ↔ 加权 DRF；dominant_share ↔ 决定「下一轮 offer 给谁」的排序键；终止条件「所有用户都被阻塞」就是 DRF 的稳态，一旦有任务结束就必须重跑循环（真实系统的动态再平衡）。

【与理论的对应】

drf_policy 对应算法 21.3.5 的 DRF-Greedy（cands = 未阻塞集合 $E$，min(key=dominant_share) = 第 6 行）；fluid_drf 对应 DRF-Fluid（sigma 即 $\sigma$，返回 $t_i=\sigma w_i/d_i$ 与 $x_i=t_iD_i$）。
实验 1 验证讲义数值例子：$A,B$ 的主导份额都收敛到 $2/3$（$A=3$ 任务、$B=2$ 任务），离散贪心与流体解一致。
实验 3 验证公理 3/2/4：流体解嫉妒对为 0（400 组随机配置），离散贪心因粒度产生大量嫉妒对（复现 21.3.5 的边界）；576 种谎报无一获益；$\sigma$ 提高 1% 即不可行。
公理 1（共享激励）体现在实验 2 的 min_tasks 一列：DRF 下每人都拿到 ≥3 个任务（无人饿死），而 FIFO 下三个用户是 0——这正是共享激励要防住的失败模式。
代码 21.4.2：三种流处理容错机制的模拟（Storm XOR ack / Spark 血缘 / Flink barrier）

#!/usr/bin/env python3
"""流处理三种容错机制的模拟与对比。
(a) Storm : tuple 级 XOR acking + 超时重放            -> at-least-once
(b) Spark : RDD 血缘 + 分区级并行重算（含血缘截断）    -> exactly-once（幂等/原子输出）
(c) Flink : Chandy-Lamport barrier checkpoint（对齐）  -> exactly-once
仅用标准库，随机种子固定，直接 python3 stream_fault_tolerance.py 运行。
"""
import random

N = 12          # 每个 interval 的记录数
PARTS = 3       # 源分区数
TIMEOUT = 2     # ack 超时（轮）


def h4(x):
    return f"0x{x & 0xFFFF:04x}"


# ==================== (a) Storm：XOR acking ====================
def part_a_storm(crash_record=7, trace_root="r0"):
    random.seed(7)
    ackval, traced, sink = {}, [], []
    stats = {"create": 0, "ack": 0}

    def create(root, tag):
        tid = random.getrandbits(64)
        before = ackval.get(root, 0)
        ackval[root] = before ^ tid
        stats["create"] += 1
        if root == trace_root:
            traced.append((f"create {tag}", before, ackval[root], tid))
        return tid

    def ack(root, tid, tag):
        before = ackval[root]
        ackval[root] = before ^ tid
        stats["ack"] += 1
        if root == trace_root:
            traced.append((f"ack    {tag}", before, ackval[root], tid))
        return ackval[root] == 0

    def tree(root, rec, crash=False):
        t0 = create(root, "T0")             # spout 发出根 tuple
        t1 = create(root, "T1")             # op1 发出子 tuple（锚定 T0）
        ack(root, t0, "T0")                 # op1 处理完 T0
        t2 = create(root, "T2")             # op2 发出子 tuple（锚定 T1）
        sink.append(rec)                    # op2 已把结果交给 sink
        if crash:                           # 崩溃：结果已出，但 T1/T2 从未 ack
            return False
        ack(root, t1, "T1")
        ack(root, t2, "T2")                 # sink 处理完 T2
        return True

    done = replayed = dup = 0
    latency = {}
    for rec in range(N):
        root = f"r{rec}"
        if rec == crash_record:
            tree(root, rec, crash=True)
            assert ackval[root] != 0                 # 校验值不归零 = 检测到未完成
            latency[root] = 3 + TIMEOUT
            tree(f"r{rec}'", rec)                    # 超时后 spout 重放整棵树
            replayed, dup = replayed + 1, dup + 1
        else:
            assert tree(root, rec) and ackval[root] == 0
            done, latency[root] = done + 1, 3
    print("  拓扑: spout(T0) --anchored--> bolt1(T1) --anchored--> bolt2(T2) --> sink")
    print(f"  完成树={done}  重放树={replayed}  create={stats['create']}  ack={stats['ack']}")
    print(f"  sink 输出（去重前）: {sorted(sink)}  其中重复 {dup} 条 -> at-least-once")
    print(f"  崩溃树 {f'r{crash_record}'} 的校验值={h4(ackval[f'r{crash_record}'])} != 0 "
          f"-> 超时 {TIMEOUT} 轮后判定失败；该记录端到端延迟 {latency[f'r{crash_record}']} 轮 vs 正常 3 轮")
    print(f"  校验值演算 mask16  root={trace_root}:")
    for tag, before, after, tid in traced:
        print(f"    {tag}  ^ {h4(tid)}: {h4(before)} -> {h4(after)}")
    return 1, dup


# ==================== (b) Spark：血缘 + 分区级重算 ====================
def part_b_spark(crash_interval=27, ckpt_every=5):
    per_part = N // PARTS
    print("  血缘: src[p0,p1,p2] --map(窄依赖)--> map_i[p0,p1,p2] "
          "--reduceByKey(宽依赖/shuffle)--> state_i[0]")
    lost_records = per_part                                   # 丢失的分区只需读回它自己的源数据
    last_ckpt = (crash_interval // ckpt_every) * ckpt_every
    replay_intervals = crash_interval - last_ckpt
    replay_records = replay_intervals * N
    print(f"  故障: interval {crash_interval} 时某节点崩溃，丢失 map_{crash_interval}[1] 与 state_{crash_interval}")
    print(f"  重算: map_{crash_interval}[1] <- 源分区1 的 {lost_records} 条（各分区并行，1 轮）")
    print(f"        state_{crash_interval} <- 检查点 interval {last_ckpt} 起重放 "
          f"{replay_intervals} 个 interval = {replay_records} 条（1 轮）")
    print(f"  总计重算 {lost_records + replay_records} 条记录，额外延迟 2 轮；重复输出 0"
          f"（未提交的批次整体重算，提交是原子的）")
    print("  检查点间隔 vs 恢复代价（interval=12 条记录，故障发生在 interval 27）：")
    for ck in (1, 2, 5, 10, 20):
        ck_records = (crash_interval - (crash_interval // ck) * ck) * N
        print(f"    每 {ck:2d} 个 interval 做检查点 -> 需重放 {ck_records:3d} 条 "
              f"（血缘回溯深度 {crash_interval - (crash_interval // ck) * ck}）")
    lam = 0.8
    print(f"  并行恢复 vs 单点 upstream backup（lambda={lam}, 检查点 1 个 interval 前）: "
          f"t_up={lam/(1-lam):.1f}  " + "  ".join(f"N={n}:{lam/(n*(1-lam)):.2f}" for n in (5, 10, 20)))
    return lost_records + replay_records, 0


# ==================== (c) Flink：barrier checkpoint ====================
def part_c_flink(bar_a=4, bar_b=6, steps=8, crash_after_ckpt=True):
    """barrier 就是 Lecture 12（Chandy-Lamport 快照）里的 marker：
    源端在记录流中注入 marker，marker 沿 DAG 流动，算子收齐所有输入的 marker 后快照自身状态。"""
    rows = {"A": [], "B": []}
    state, buffered = 0, []
    aligned, snapshot, offset = False, None, None
    assert bar_a < bar_b
    for s in range(1, steps + 1):
        if s < bar_a:                                     # A 上的快照前记录
            rows["A"].append(".")
            state += 1
        elif s == bar_a:
            rows["A"].append("|")                         # A 的 barrier 先到：停读 A
        elif not aligned:
            rows["A"].append("x")                         # 缓存 A 的后继记录，等 B 的 barrier
            buffered.append(f"A{s}")
        else:
            rows["A"].append(".")
            state += 1
        if s < bar_b:
            rows["B"].append(".")
            state += 1
        elif s == bar_b:
            rows["B"].append("|")
            aligned = True                                # 收齐两条通道的 barrier -> 对齐完成
            snapshot, offset = state, {"A": bar_a, "B": bar_b}
            state += len(buffered)                        # 缓存记录属于快照前状态，快照后处理
        else:
            rows["B"].append(".")
            state += 1
    print("  barrier 对齐（. = 已处理, | = barrier, x = 缓存等对齐）")
    print("        step:  " + "  ".join(f"{s}" for s in range(1, steps + 1)))
    for ch in ("A", "B"):
        print(f"    通道{ch}:      " + "  ".join(rows[ch]))
    print(f"  对齐期缓存记录 {buffered}（共 {len(buffered)} 条）-> 这就是 aligned checkpoint 的停顿来源")
    print(f"  快照: state={snapshot}, 源 offset={offset}（barrier 位置 = 可重放起点）")
    replayed = len(buffered) if crash_after_ckpt else 0
    dup = 0 if crash_after_ckpt else 1
    print(f"  故障: 检查点完成后 op2 崩溃 -> 回滚到 state={snapshot}、offset={offset}，重放 {replayed} 条；"
          f"重复输出 {dup}")
    return replayed, dup


print("=" * 78)
print("(a) Storm：tuple 级 XOR acking（校验值演算 + 超时重放）")
print("=" * 78)
storm_replay, storm_dup = part_a_storm()
print("=" * 78)
print("(b) Spark：RDD 血缘、分区级重算与血缘截断")
print("=" * 78)
spark_replay, spark_dup = part_b_spark()
print("=" * 78)
print("(c) Flink：barrier checkpoint 的对齐过程")
print("=" * 78)
flink_replay, flink_dup = part_c_flink()
print("=" * 78)
print("  机制            重算/重放记录数   重复输出   语义")
print(f"  Storm           {storm_replay:<16d} 有({storm_dup})       at-least-once")
print(f"  Spark 血缘      {spark_replay:<16d} 无({spark_dup})       exactly-once")
print(f"  Flink barrier   {flink_replay:<16d} 无({flink_dup})       exactly-once")
print("=" * 78)

【代码做什么？】

(a) Storm XOR acking：为每条记录建三层 tuple 树（spout $T_0$ → bolt1 $T_1$ → bolt2 $T_2$ → sink）。create 把新 tuple 的 64 位随机 id 异或进校验值、ack 把它异或出；crash_record=7 的那棵树在「已发出结果但未 ack 输入」时崩溃，代码断言此时校验值 $\ne 0$（这就是失败检测依据），随后打印超时与重放，并统计出 sink 收到 13 条而实际只有 12 条（1 条重复）——at-least-once 的直接证据；还打印 r0 的逐事件校验值演算，末步 0xa450 ^ 0xa450 = 0x0000 归零。
(b) Spark 血缘重算：打印血缘图、故障点、重算计划（丢失分区只需读回自己的源分区 4 条；有状态算子从检查点重放 24 条）、检查点间隔 vs 重放记录数对照表（每 1/2/5/10/20 个 interval ⇒ 重放 0/12/24/84/84 条），并用论文模型打印并行恢复 vs 单点 upstream backup（$\lambda=0.8$ 时 4.0 vs 0.80/0.40/0.20）。
(c) Flink barrier checkpoint：两条输入通道分别在 step 4 与 step 6 收到 barrier；代码按真实规则停读先到 barrier 的通道并缓存其后续记录，待另一条通道的 barrier 到达后对齐 → 快照 → 广播 barrier → 处理缓存记录，并打印对齐过程 ASCII 图（. 已处理 / \| barrier / x 缓存等待）。
汇总：打印三种机制的对照（重算/重放记录数、是否有重复输出、达成的语义）。

【分布式机制透视】

ackval 字典就是 acker 的状态，create/ack 就是 bolt 与 acker 之间的记账消息（每 tuple 2 条）；sink 列表就是外部输出，其中的重复项正是 at-least-once 的脏数据。
lineage 字典是 driver 侧的 RDD 依赖元数据，重算 = 「按血缘重新调度任务」；state/snapshot 是算子状态与一致快照，offset 是源的可重放位置，而「缓存等待」就是对齐停顿的真实来源。
三种故障点刻意不同：(a) 在「输出已产生、确认未到达」时崩溃（最坏情形，制造重复）；(b) 在「节点丢失内存分区」处崩溃（考察重算）；(c) 在「检查点完成之后」崩溃（考察回滚重放）。对比这三种处理方式，就理解了语义差异的根源。
「分布式」体现在：树的分支由不同 bolt 并行处理、ack 乱序到达（故需交换律）；血缘跨 stage 与多机（故恢复是分区级并行）；barrier 在多通道上赛跑（故需对齐）。

【与理论的对应】

(a) 的 create/ack 两行异或对应算法 21.3.1 的不变量 (I) $V_r=\bigoplus_{t\in P_r}\mathrm{id}(t)$：「崩溃后 $V\ne0$」验证检错性，「完成树 $V=0$」是不变量在 $P_r=\emptyset$ 时的特例。
(b) 对应算法 21.3.2 的执行与恢复阶段（每一步就是「按依赖再算一遍」，即血缘重建定理的归纳步）；对照表对应 lineage cutoff 的代价权衡。
(c) 逐行对应算法 21.3.4，也对应 Lecture 12 的 Chandy-Lamport：barrier = marker、快照 = 记录本地状态、广播 = 沿输出通道传播 marker、offset = 通道状态的重放点。
三者的重复输出统计（有/无/无）印证 21.2.2 的语义表：XOR ack 只给 at-least-once；原子提交的批与一致快照 + 2PC 才给出 exactly-once。
代码 21.4.3：集群调度策略仿真对比（FIFO / Fair / DRF）

#!/usr/bin/env python3
"""集群调度策略对比：FIFO / Fair Sharing（单资源 CPU 公平）/ DRF。
离散时间步模拟：作业按到达时间进入队列，每个 tick 调度器决定启动哪些任务。
输出：平均完成时间、平均排队时间、资源利用率、Jain 公平指数、利用率轨迹（看队头阻塞）。
仅用标准库，随机种子固定，直接 python3 scheduler_compare.py 运行。
"""
import random

CPU, MEM = 32.0, 64.0
CAP = {"cpu": CPU, "mem": MEM}
RES = ("cpu", "mem")
TICKS = 80


class Job:
    def __init__(self, jid, arrival, ntasks, demand, duration):
        self.jid, self.arrival, self.demand = jid, arrival, dict(zip(RES, demand))
        self.pending, self.duration = ntasks, duration
        self.running = []                      # 每个元素 = 剩余 tick 数
        self.used = {r: 0.0 for r in RES}
        self.share_sum, self.active_ticks = 0.0, 0
        self.start = self.finish = None

    def fits(self, free):
        return all(free[r] >= self.demand[r] for r in RES)

    def dominant(self):                        # 当前主导份额
        return max(self.used[r] / CAP[r] for r in RES)

    def start_task(self, free):
        for r in RES:
            free[r] -= self.demand[r]
            self.used[r] += self.demand[r]
        self.running.append(self.duration)
        self.pending -= 1
        if self.start is None:
            self.start = NOW

    def tick(self):
        done = 0
        for i in range(len(self.running) - 1, -1, -1):
            self.running[i] -= 1
            if self.running[i] == 0:
                self.running.pop(i)
                done += 1
        return done

    def release(self, done, free):
        for r in RES:
            self.used[r] -= done * self.demand[r]
            free[r] += done * self.demand[r]


def build_jobs():
    random.seed(11)
    jobs = [Job(0, 0, 12, (4, 2), 10),         # CPU 密集型长作业：12 个任务，只有 8 个能同时跑
            Job(7, 0, 4, (1, 16), 15)]         # 内存密集型：CPU 需求低、内存需求极高
    specs = [(1, 2, 4, (4, 2), 3), (2, 1, 4, (2, 4), 3), (3, 0, 6, (3, 2), 4),
             (4, 0, 1, (2, 4), 8), (5, 2, 2, (2, 4), 2), (6, 5, 1, (1, 2), 2)]
    for jid, arrival, ntasks, demand, duration in specs:
        jobs.append(Job(jid, arrival, ntasks, demand, duration))
    return jobs


def pick_fifo(jobs, free):
    for j in jobs:                              # 严格按到达顺序：只看队头
        if j.arrival <= NOW and j.pending > 0:
            return j if j.fits(free) else None   # 队头装不下 -> 后面全部饿死
    return None


def pick_fair(jobs, free):
    cands = [j for j in jobs if j.arrival <= NOW and j.pending > 0 and j.fits(free)]
    return min(cands, key=lambda j: (j.used["cpu"], j.jid)) if cands else None


def pick_drf(jobs, free):
    cands = [j for j in jobs if j.arrival <= NOW and j.pending > 0 and j.fits(free)]
    return min(cands, key=lambda j: (j.dominant(), j.jid)) if cands else None


def jain(xs):
    xs = list(xs)
    s = sum(xs)
    return s * s / (len(xs) * sum(v * v for v in xs)) if s > 0 else 0.0


def simulate(policy):
    global NOW
    jobs = build_jobs()
    free = {r: CAP[r] for r in RES}
    busy = {r: 0.0 for r in RES}
    trace = []
    for t in range(TICKS):
        NOW = t
        for j in jobs:                          # 1. 完成的任务释放资源
            j.release(j.tick(), free)
        for j in jobs:                          # 2. 记录活跃作业的瞬时主导份额
            if j.arrival <= t and (j.pending > 0 or j.running):
                j.share_sum += j.dominant()
                j.active_ticks += 1
        for j in jobs:                          # 3. 调度：反复挑选直到装不下
            if j.pending == 0 and not j.running and j.start is not None and j.finish is None:
                j.finish = t
        while True:
            j = policy(jobs, free)
            if j is None:
                break
            j.start_task(free)
        for r in RES:                           # 4. 本 tick 的利用率
            busy[r] += CAP[r] - free[r]
        trace.append(sum((CAP[r] - free[r]) / CAP[r] for r in RES) / len(RES))
    return jobs, busy, trace


print("=" * 92)
print("作业集（到达时间 / 任务数 / 每任务需求 / 每任务时长）")
print("=" * 92)
for j in build_jobs():
    print(f"  J{j.jid}: arrival={j.arrival}  tasks={j.pending:2d}  demand={tuple(j.demand[r] for r in RES)}"
          f"  duration={j.duration}  总需求={tuple(j.pending * j.demand[r] for r in RES)}")

results = {}
for name, pol in (("FIFO", pick_fifo), ("Fair(CPU)", pick_fair), ("DRF", pick_drf)):
    results[name] = simulate(pol)

print("=" * 92)
print("各策略的作业级结果（start/finish 为 tick 编号；wait = start - arrival）")
print("=" * 92)
for name, (jobs, busy, trace) in results.items():
    print(f"  [{name}]")
    for j in jobs:
        st = "-" if j.start is None else str(j.start)
        fi = "-" if j.finish is None else str(j.finish)
        print(f"    J{j.jid}: start={st:>3s} finish={fi:>3s} wait={'-' if j.start is None else j.start - j.arrival:>3}"
              f" 未启动任务={j.pending}")

print("=" * 92)
print("汇总对比")
print("=" * 92)
print(f"  {'策略':10s} {'平均完成时间':>12s} {'平均排队时间':>12s} {'makespan':>8s} {'CPU利用率':>10s} "
      f"{'MEM利用率':>10s} {'Jain(份额)':>10s} {'完成作业数':>10s}")
for name, (jobs, busy, trace) in results.items():
    fin = [j.finish - j.arrival for j in jobs if j.finish is not None]
    wait = [j.start - j.arrival for j in jobs if j.start is not None]
    assert fin, "no job finished"
    shares = [j.share_sum / j.active_ticks if j.active_ticks else 0.0 for j in jobs]
    span = max(j.finish for j in jobs if j.finish is not None)
    print(f"  {name:10s} {sum(fin)/len(fin):12.1f} {sum(wait)/len(wait):12.1f} {span:8d} "
          f"{busy['cpu']/(CAP['cpu']*span)*100:9.1f}% {busy['mem']/(CAP['mem']*span)*100:9.1f}% "
          f"{jain(shares):10.3f} {len(fin):10d}")

print("=" * 92)
print("利用率轨迹（每字符 = 1 tick 的平均资源利用率，0-9 = 0%-90%+；FIFO 的 '2' 段就是队头阻塞）")
print("=" * 92)
for name, (jobs, busy, trace) in results.items():
    span = max(j.finish for j in jobs if j.finish is not None)
    line = "".join(str(min(9, int(v * 10))) for v in trace[:span + 2])
    print(f"  {name:10s} {line}")
print("  " + " " * 11 + "".join(str(t // 10) if t % 10 == 0 else " " for t in range(TICKS)))
print("=" * 92)
print("  关键观察: FIFO 让队头长作业 J0 压住队尾；Fair(CPU) 只看 CPU, 会被内存密集型 J7 吃掉内存,"
      " 导致 CPU 闲置；DRF 用主导份额同时盯住两种资源")
print("=" * 92)

【代码做什么？】

异构作业集（固定种子）：J0 是 CPU 密集长作业（12 个任务 × (4 CPU, 2 GB)，每个跑 10 个 tick），J7 是内存密集作业（4 个任务 × (1 CPU, 16 GB)，15 个 tick），另加 6 个到达时间与需求各异的中小作业；集群容量 (32 CPU, 64 GB)。
三种调度器：pick_fifo（严格按到达顺序，只看队头：队头装不下则无人能跑 ⇒ 队头阻塞）、pick_fair（在装得下的作业里挑 CPU 占用最小者）、pick_drf（挑 主导份额最小者）。
离散时间步仿真：每个 tick 依次释放已完成任务的资源、记录活跃作业的瞬时主导份额、标记完成的作业、按策略反复启动任务直到装不下，并累计本 tick 的资源使用量。
输出：每个作业的 start/finish/wait 与未启动任务数；汇总表的平均完成时间、平均排队时间、makespan、CPU 利用率、内存利用率、Jain 公平指数、完成作业数；最后是利用率轨迹图（每字符 = 1 tick 的平均资源利用率）。

【分布式机制透视】

决策有时间维度：与代码 21.4.1 的静态一次性分配不同，这里是滚动决策（任务完成 → 释放 → 再决策），正是 YARN 心跳、K8s 调度循环、Mesos offer 轮次的运行方式。静态 DRF 只回答「份额应该是多少」，滚动调度器必须每秒回答「现在给谁」——所以真实系统用的是瞬时主导份额。
队头阻塞的形式化：return j if j.fits(free) else None 这一行就是它——队头作业装不下时，后面即使资源足够也不被考虑；轨迹图里 FIFO 前 20 个 tick 稳定在 6–7，就是在跑 J0 而其余作业全部饿死。
多资源失效：pick_fair 只看 CPU，于是内存密集的 J7 被一路喂到「CPU 份额达标」（拿到 3 个任务 = 48 GB），把内存吃干，CPU 密集作业无处可放 ⇒ CPU 利用率被拖低；DRF 看到 J7 的内存份额已超 CPU 份额便自动限速，把内存留给 CPU 密集作业。
映射：Tick ↔ 调度周期；demand ↔ Pod requests；pick_drf ↔ Mesos 的 DRF 排序；shares 统计 ↔ 公平性监控；Jain ↔ 多租户公平性度量。

【与理论的对应】

pick_drf 就是算法 21.3.5 的「选主导份额最小者」；pick_fair 是同一框架下只把 $key$ 换成单资源份额，从而在实验上隔离出「单资源公平 vs 多资源公平」的差异（DRF 动机的定量验证）。
结果与理论一致：DRF 三项指标同时最好（平均完成时间 13.6 < 16.8 < 28.6 tick，CPU 利用率 74.0% > 65.3% > 61.6%，Jain 0.457 > 0.414 > 0.291），FIFO 平均排队时间高达 19.9（队头阻塞），单资源公平在需求异构时会把 CPU 闲置。
同时印证 DRF 的边界：它不追求每种资源都用满（仿真中内存利用率 71.5%），保证的是主导份额公平 + 帕累托效率；提高利用率要靠回填、超额分配、装箱等另一层机制。

21.5 性能与可扩展性分析

Storm：tuple 级 ack 的固定成本。每个 tuple 产生 2 条记账消息（create + ack），即 $O(\text{树的 tuple 数})$——当 tuple 很小时，记账流量与数据流量同量级，吞吐被记账吃掉。D-Streams 论文的实测正是证据：30 节点、100 字节记录下，Spark Streaming 的 Grep 达 670K records/s/节点，Storm 只有 115K（差约 5.8×；论文还说明已为 Storm 做过「每 100 条批量更新」「reduce 每秒只发一次计数」等优化），1000 字节记录时 Storm 变好但仍慢约 2×；同场景的 S4 每节点最多约 7500 records/s（慢近一个数量级）。逐记录确认是「用吞吐买语义」的典型交易，Heron 重新设计 acking 并引入 TCP/spout/逐级背压正是为了降低这笔成本。
Storm 的可扩展性：parallelism hint 提高并行度即可扩容，但瓶颈会转移——fields grouping 的热点 key（同一个大 V 的推文都落到同一 task）与 acker 热点（一棵树的所有记账消息都发给同一个 acker）。大规模拓扑常用多级聚合（task 内先局部聚合，再按 key 分区做全局聚合）。
Spark 的内存依赖与退化：cache 是性能来源，也是风险来源——内存不足时分区 spill 到磁盘，性能退化到 MapReduce 量级（甚至更差，因为多了序列化与 GC 开销）；MEMORY_ONLY 级别下分区被驱逐后还要按血缘重算。D-Streams 论文特别指出：Spark Streaming 的任务只有 50–200 ms，对 GC 停顿尤其敏感。工程缓解：选合适的 StorageLevel、用 reduceByKey 而非 groupByKey（map 侧先本地聚合，大幅减少 shuffle）、增大分区数、避免 collect() 把数据拉回 driver。血缘元数据本身很小（$O(V+E)$），但它决定的链长就是恢复时间。
Spark Streaming 的延迟下限：端到端延迟下界 ≈ 批间隔 + 本批处理时间，这是结构性的，加机器无法消除。论文的配置是「1 秒延迟目标 ⇒ 500 ms 批间隔；2 秒目标 ⇒ 1 秒批间隔」；规模结果为 100 节点上 Grep 达 6 GB/s（约 64M records/s）且亚秒级延迟，CPU 密集的 WordCount/TopK 约 2.3 GB/s（25M records/s）。让时间步「不漏气」的两个优化是 timestep pipelining（下一个时间步的任务可在当前步未结束时提交） 与异步网络 IO。
微批的恢复表现（论文实测）：1 秒批、20 节点、WordCount 下，检查点间隔 10 秒时单次故障最多带来约 1 秒的处理延迟；间隔 2 秒时恢复仅需 0.15 秒；间隔放宽到 30 秒则延迟达 3.5 秒、约 18 秒才重新稳定。节点数 20→40 时恢复时间约减半，直接验证了「并行恢复优于单点串行追赶」（$t_{par}\approx\lambda/(N(1-\lambda))$）。论文还用推测执行处理慢节点（straggler）：任务比同 stage 中位任务慢 1.4× 即启动副本，把 3.02 s 的 interval 拉回 1.00 s——在确定性微批模型里推测执行几乎免费。
Flink 的对齐停顿与背压：对齐式 checkpoint 的停顿来自「先到 barrier 的通道要等其它通道」。在背压下，慢通道的数据在缓冲里排队、barrier 也排在队尾，于是(a) 对齐时间 ∝ 慢通道积压量、(b) 形成「暂停—突发」的延迟尖峰（p99 显著抬高）、(c) 缓存本身加剧背压（正反馈）。非对齐 checkpoint（unaligned）让 barrier 越过队列中的记录（把在途数据当作通道状态写进快照），几乎消除停顿，代价是快照变大 + 额外落盘 IO。工程选择：状态大、背压重 ⇒ unaligned；状态中等、延迟平稳 ⇒ aligned。检查点间隔越小恢复越快，但检查点自身的 IO 与对齐停顿会周期性影响吞吐与尾延迟。

容错机制	稳态开销	恢复时间	恢复粒度	能容忍的故障	主要瓶颈
Storm：tuple 级 XOR ack	每 tuple 2 条记账消息（与数据量同阶）	超时 + 重放整棵树（秒级）	整棵 tuple tree	任意节点	ack 消息开销、acker 热点
Spark：血缘 + 检查点	近似为 0（无复制/落盘），检查点有周期成本	分区级并行重算（亚秒~秒级）	分区级	任意 worker	血缘链长度、内存不足导致 spill
Flink：barrier checkpoint	每条边 1 条 barrier + 状态快照 IO	回滚到最近检查点 + 重放	算子状态 + 输入偏移（一致割）	任意算子（受最近已完成检查点限制）	对齐停顿（背压下明显）、快照大小
上游备份（对照）	上游缓冲消息（内存压力大）	单点串行追赶 $t_{up}=\lambda/(1-\lambda)$（秒~分钟）	单节点状态	单节点	恢复慢、无法处理 straggler
热备复制（对照）	2× 硬件 + 同步协议	快（毫秒级接管）	副本状态	副本不能同时故障	成本翻倍、协议复杂

DRF 的调度开销与大规模适用性：算法极便宜（流体版 $O(nm)$、贪心版 $O(nm\sum_i t_i)$，而用户数与资源类型数都只有几十/几种），DRF 从来不是调度吞吐的瓶颈。真正的瓶颈是决策往返与状态同步：Mesos 每次 offer 都要一次网络往返且框架可能拒绝/超时，粒度越细、框架越多，空转越大；Omega 用共享状态 + 乐观并发（调度器直接读写完整集群状态，冲突检测后重试）把串行中央队列变成并行乐观事务，代价是冲突时浪费调度工作；Kubernetes 走集中式 + 工程优化路线（Scheduler 的 filter + score 两阶段，配合亲和/反亲和、污点与容忍、requests/limits、QoS、抢占，并用调度器缓存与乐观并发/分片缓解扩展压力）；Borg 的实践则证明单一集中式调度器 + 优先级 + 抢占 + gang 调度在 Google 规模可行——修正了「共享状态一定优于集中式」的直觉。抢占提高利用率与关键作业 SLA，但被抢占任务已消耗的算力全部作废，因此要「杀最年轻的任务」（Fair Scheduler 做法）、偏好可重算的批任务、并限制频率避免抖动。

目标	提升手段	代价	典型体现
高利用率	装箱/回填、超额分配、抢占、细粒度共享	公平性下降、突发无余量、抢占重算	Mesos 细粒度共享、Borg overcommit
公平性	DRF、Fair Scheduler 等份额、延迟调度	利用率下降（份额小者装不满时资源闲置）	DRF 的「资源未用满」
低延迟	小批间隔、逐记录、非对齐 checkpoint、小任务粒度	固定开销上升、快照变大、对齐停顿	微批的批间隔下限
数据局部性	延迟调度、按 key 分区复用	短暂空转、错过公平份额	Hadoop 延迟调度、Spark `partitionBy`

两条主线的结论：(1) 流处理的性能几乎总由容错机制的开销决定——逐记录确认把吞吐压掉一半以上（实测 2–6×），微批把延迟钉在批间隔以上，barrier 对齐在背压下制造尾延迟尖峰。没有免费的语义。 (2) 集群调度的瓶颈不在算法（DRF 只有 $O(nm)$），而在往返、状态一致性与抢占浪费；因此其演化方向是更少的往返（offer 批量化）、更少的冲突（乐观并发）、更强的约束表达（filter+score）与更可控的抢占。

21.6 关键要点

流处理的黄金法则：把「容错」从「复制数据」变成「重算数据」——Storm 用 tuple 级 XOR ack 定位未完成的树再重放，Spark/D-Streams 用血缘做分区级重算，Flink 用 barrier（= Chandy-Lamport 的 marker）拍一致快照再重放。这与 MapReduce 的 re-execution 是同一种哲学，只是粒度更细、延迟更低，且不需要热备复制的 2× 硬件。
语义、延迟、吞吐三者不可兼得：at-least-once 便宜（XOR ack）、exactly-once 昂贵（快照 + 2PC/幂等输出）、微批把延迟下限钉在批间隔、逐记录确认把吞吐吃掉 2–6×。「恰好一次」是「效果的恰好一次」，要求输入可重放 ⊕ 状态可回滚 ⊕ 输出幂等/事务三条同时成立。
XOR acking 的精髓是用一个 8 字节不变量代替整棵树的账本：$V_r=\bigoplus_{t\in P_r}\mathrm{id}(t)$，创建时异或进来、完成时异或出去；因为异或可交换、可结合、自逆，ack 可任意顺序到达且每个 spout tuple 的状态是 $O(1)$。两条硬约束：每个 tuple 必须恰好 ack 一次（重复 ack 会让校验值无法归零，极小概率下还会提前归零酿成丢数据），且 id 需唯一（碰撞/随机归零概率约 $2^{-64}$）。
Spark 快在「内存 + 血缘」，慢也慢在这两处：cache 让迭代/交互负载快一个数量级，但内存不足就 spill 到磁盘、退化到 MapReduce 量级；血缘让恢复无需复制数据，但链越长恢复越慢 ⇒ 必须用检查点做血缘截断——「重算」与「检查点」是一对必须同时出现的机制。
调度的黄金法则：多资源公平 = 在「主导份额」这个标量上做 max-min 公平。规则只有一句——始终把下一份资源给主导份额最小的用户——却同时满足共享激励、策略防伪、无嫉妒、帕累托效率四条公理。
DRF 的边界：四条公理建立在资源可无限细分 + 任务同构 + 无放置约束 + 无时长/优先级之上；真实系统（YARN/K8s/Mesos）都在其上叠加抢占、优先级、配额与约束求解，且 DRF 只保证帕累托效率，不保证每种资源用满（18 CPU/36 GB 的例子用满 CPU 却空出 8 GB 内存）。

21.7 常见陷阱与注意事项

把 exactly-once 当成「记录只投递一次」，于是认为「Storm 加了 ack 就是 exactly-once」。ack 只保证至少一次：重放会让下游重复处理、输出重复。正确做法是接受 at-least-once 并让输出幂等（按业务键去重），或上 Trident/事务性拓扑，或走 Spark Streaming 的「可重放输入 + 原子提交输出」与 Flink 的「快照 + 2PC sink」。
对有状态聚合用 shuffle grouping。shuffle 是轮询分发，同一个 key 会落到不同 task，每个 task 只算出局部计数——结果错了而且很隐蔽（每个 task 单独看都正常）。正确做法：有状态聚合必须用 fields grouping（按聚合键哈希），需要全局视图才用 all grouping（广播，注意流量是并行度的倍数）。
漏 ack 或重复 ack。漏 ack ⇒ 校验值永不归零 ⇒ 一直超时重放（内存泄漏 + 重复处理，讲义明确警告「每个 tuple 都占用内存」）；重复 ack ⇒ 同一个 id 被抵消两次 ⇒ 该 tuple 在统计上「永远 pending」⇒ 校验值无法归零 ⇒ 该树判失败并重放（多余的重复处理），极小概率下会提前归零 ⇒ 静默丢数据。正确做法：每个 tuple 在恰好一条路径上 ack 或 fail，用 try/finally 覆盖异常路径，并监控「超时重放率」。
认为「校验值归零 ⟺ 树完成」是绝对等价。它是概率意义上的等价：若同时在途的 id 恰好满足某种异或关系，或发生 id 碰撞，就会假阳性。正确做法：记住这是 $O(2^{-64})$ 量级的工程近似，并保证 id 唯一与「恰好一次 ack」两条硬前提。
认为「有了血缘就不需要检查点」。血缘只保证可重算，不保证重算得起：有状态算子的血缘链会无限延伸，故障时可能要回放很久。正确做法：周期性 checkpoint + lineage cutoff，并按「恢复时间目标 vs 稳态开销」选间隔（论文：2 秒检查点 ⇒ 0.15 秒恢复；30 秒 ⇒ 3.5 秒延迟）。
把 watermark 当成「数据已经到齐」的证明。它只是「我不再等了」的承诺，且依赖源端不乱发迟到数据；真正的迟到数据要靠 allowed lateness + 触发器 + 结果更新（或 D-Streams 的「松弛时间 + 应用层修正」）处理。正确做法：把 watermark 理解为延迟与完整性的旋钮，并为迟到数据定义补偿逻辑。
认为「DRF 最优 ⇒ 资源一定被用满」。DRF 保证帕累托效率，不保证利用率最大：$C=(18,36)$ 的 DRF 解用满 CPU（18/18）却只用 28/36 内存，剩下的 8 GB 谁也拿不走（再给谁一个任务都需要 CPU）。正确做法：把利用率交给回填、超额分配、装箱、抢占等另一层机制。
认为「DRF 无嫉妒对任何情况都成立」，或把背压当成故障。两条边界：(a) 加权 DRF 下朴素无嫉妒不成立（应按 $w_i/w_j$ 折算），(b) 离散粒度下贪心解会偏离流体解（$(10,10)$、$(1,1)$ 与 $(9,9)$ 的反例）。而背压是正常的流控机制，真正要警惕的是它与 checkpoint 的相互作用——对齐式 checkpoint 在持续背压下会把停顿放大成延迟尖峰，必要时切 unaligned 或调整并行度/批间隔。

21.8 思考题（带答案）

题 1（计算题，DRF）：集群 $C=(\text{120 CPU},\ \text{36 GB})$；Job 1 每任务 $( ext{2 CPU}, ext{8 GB})$，Job 2 每任务 $( ext{6 CPU}, ext{2 GB})$。(1) 指出两者的主导资源；(2) 求 DRF 的公平解（各多少任务、主导份额、资源用量）；(3) 若集群改为 $C=(\text{18 CPU},\text{36 GB})$，解变成什么？为什么？

答：(1) $d_1=\max(2/120,\ 8/36)=\max(0.0167,0.2222)=0.2222$ ⇒ Job 1 内存主导；$d_2=\max(6/120,\ 2/36)=\max(0.05,0.0556)=0.0556$ ⇒ Job 2 也是内存主导（CPU 太便宜，$6/120<2/36$，主导资源发生「翻转」——这正是讲义思考题的答案）。 (2) $\sigma=\min\left(\frac{120}{2/0.2222+6/0.0556},\ \frac{36}{8/0.2222+2/0.0556}\right)=\min\left(\frac{120}{117},\frac{36}{72}\right)=0.5$，于是 $t_1=\frac{0.5}{0.2222}=2.25$、$t_2=\frac{0.5}{0.0556}=9$：Job 1 得 2.25 个任务（4.5 CPU、18 GB），Job 2 得 9 个任务（54 CPU、18 GB），两者主导份额都是 $0.5$；内存 36/36 = 100%，CPU 58.5/120 = 48.75%（内存是瓶颈）。 (3) 换成 $(18,36)$：$d_1=\max(1/9,2/9)=2/9$（内存主导），$d_2=\max(6/18,1/18)=1/3$（CPU 主导）；$\sigma=\min\left(\frac{18}{9+18},\frac{36}{36+6}\right)=\min(0.667,0.857)=2/3$ ⇒ $t_1=3$、$t_2=2$，两者份额都是 $2/3$（讲义原例）。根因：「谁是主导资源」取决于需求与容量的比值——CPU 从 120 降到 18，Job 2 的 CPU 占比从 $1/20$ 升到 $1/3$ 而成为主导资源。同一组作业，资源配比一变，「公平」的具体分配就完全不同，这正是 DRF 优于「按某一固定资源平均分」的地方。

题 2（推演题，Storm XOR acking）：树结构为 $T_0$(spout, 0xA001) → $T_1$(0xB002) → $T_2$(0xC003) → $T_3$(0xD004)；事件顺序为 create T0、create T1、ack T0、create T2、ack T1、create T3、ack T2、ack T3。(1) 逐步写出校验值；(2) 树是否完成？(3) 若 $T_2$ 被 ack 两次会怎样？(4) 若事件顺序打乱，结论会变吗？

答：(1) create T0 ^A001: 0000→A001；create T1 ^B002: A001→1003；ack T0 ^A001: 1003→B002；create T2 ^C003: B002→7001；ack T1 ^B002: 7001→C003；create T3 ^D004: C003→1007；ack T2 ^C003: 1007→D004；ack T3 ^D004: D004→**0000**。 (2) 完成：$V=0$ ⇒ pending 集合为空 ⇒ 通知 spout 成功（中途未出现提前归零）。 (3) 重复 ack 使该 id 被抵消两次，等价于它「永远 pending」：ack T2(重复) → D004^C003=1007，ack T3 → 1007^D004=C003≠0 —— 树走到超时被判失败并整体重放（多余的重复处理，但仍不丢数据）；极小概率下若剩余 pending 的异或恰好等于该 id，校验值会提前归零 ⇒ acker 提前宣布成功 ⇒ 静默丢数据。可见「重复 ack」同时是「假失败（常见）」与「假成功（罕见但致命）」的 bug 来源，而「漏 ack」只造成假失败。 (4) 结论不变：异或满足交换律与结合律，$V=\bigoplus_{t\in P_r}\mathrm{id}(t)$ 只与「每个 id 被异或进来的奇偶次数」有关，与到达顺序无关——这正是 Storm 敢让不同 bolt 并行处理同一棵树不同分支的依据。会破坏结论的只有 id 碰撞与非「恰好一次」的 ack。

题 3（「直观但错误的想法」）：某同学说：「DRF 让每个用户在自己最缺的资源上都被满足到同等程度，那集群的每种资源最终都会被用满，否则就是浪费。」错在哪？

答：错在把帕累托效率与利用率最大化混为一谈。DRF 保证的是「不存在能在不损害任何人的前提下改进某人的分配」，并不要求每种资源都用尽。反例就是讲义原例：$C=(18\,\text{CPU},36\,\text{GB})$ 下 DRF 给 Job 1 三个任务、Job 2 两个任务，CPU 用满 18/18，内存只用 $3\times8+2\times2=28$ GB，空出 8 GB。空着为什么不算浪费？因为再给 Job 1 一个任务需要 2 个 CPU（已无），再给 Job 2 一个需要 6 个 CPU（已无）——瓶颈资源已打满，空闲内存无法在「不损害他人」的前提下被利用，分配仍然帕累托最优（题 1(3) 中把 $\sigma$ 提到 $2/3$ 以上即不可行也印证了这一点）。正确总结：公平与效率是两个目标，DRF 只保证前者加帕累托意义下的后者；「用满所有资源」要靠回填、装箱、超额分配等另一层机制。

题 4（「直观但错误的想法」）：某同学说：「Spark 的血缘能精确重建任何丢失分区，所以既不需要检查点，也不必担心恢复时间；Flink 的 barrier checkpoint 反正也是重算，两者没有本质差别。」错在哪？

答：三处错误。(1)「不需要检查点」错：血缘只保证可重算，重算量 = 血缘链长度；对短链无状态批作业确实可以不用检查点，但流式有状态算子的血缘会随运行时间无限延伸（论文的 WordCount 每轮还要「减去 30 秒前的数据」），没有检查点意味着故障后可能要回放几小时的数据 ⇒ 必须周期检查点 + lineage cutoff（论文实测：2 秒间隔 ⇒ 恢复 0.15 秒；30 秒 ⇒ 延迟 3.5 秒、约 18 秒才稳定）。(2)「不必担心恢复时间」错：恢复时间是可用性指标；血缘的价值恰恰在于把恢复从「一台备用机串行追赶」（$t_{up}=\lambda/(1-\lambda)=4.0$，$\lambda=0.8$）变成「全集群分区并行重算」（$N=5/10/20$ 时 0.80/0.40/0.20）——前提是恢复任务能被并行调度，检查点太稀疏或并行度不足时仍会失控。(3)「没有本质差别」错：差别在一致性来源与延迟结构——Spark Streaming 的 exactly-once 来自微批的原子提交（延迟被批间隔钉死在秒级），Flink 的来自Chandy-Lamport 一致快照 + 2PC 输出（不必等批边界，可做到亚秒/毫秒级，代价是对齐停顿与检查点复杂度）。二者的共同点只有一句：都不复制数据，而是重算数据（本章黄金法则）——但「重算」的时机、粒度与一致性论证完全不同：一个是批边界上的原子提交，一个是全局一致割上的快照。

Share on

Bluesky Facebook LinkedIn Mastodon X (formerly Twitter)

Runguo Li

第八部分：现代分布式系统与真实世界

Lecture 21: Stream Processing and Cluster Scheduling — 流处理与集群调度

21.1 概述

21.2 核心概念与分布式机制图解

21.2.1 从批处理到流处理：有界数据 vs 无界数据（Bounded vs Unbounded Data）

21.2.2 流处理的六个关键概念：事件时间、窗口、水位线、状态、背压、投递语义

21.2.3 Apache Storm 的编程模型：Topology / Spout / Bolt / Tuple / Grouping

21.2.4 Storm 的架构与可靠性机制：Nimbus / Supervisor / ZooKeeper / Anchoring / Acker

21.2.5 Apache Spark：RDD、血缘（Lineage）与 DAG 调度

21.2.6 Spark Streaming 与 DStream：微批（Micro-batch）模型

21.2.7 Flink 与 Chandy-Lamport barrier checkpoint：Lecture 12 的快照算法就在流处理的引擎里

21.2.8 集群调度的演进：从单体调度到共享状态调度

21.2.9 调度目标与策略：利用率、公平性、延迟与数据局部性

21.2.10 DRF 与主导份额（Dominant Share）：把一维公平推广到多维

21.3 算法伪代码与正确性分析

算法 21.3.1：Storm 的 XOR acking 可靠性机制（Anchoring + Acker + 超时重放）

算法 21.3.2：Spark 的血缘记录、分区重算与 stage 划分

算法 21.3.3：Spark Streaming 的微批执行与窗口聚合（含 exactly-once 的三个条件）

算法 21.3.4：Flink 式的 Chandy-Lamport barrier checkpoint（= Lecture 12 的 marker）

算法 21.3.5：DRF（Dominant Resource Fairness）完整调度算法与四公理证明

21.4 代码示例与分布式实现

代码 21.4.1：DRF 调度器完整实现（主导份额 + 逐任务分配 + 四公理验证）

代码 21.4.2：三种流处理容错机制的模拟（Storm XOR ack / Spark 血缘 / Flink barrier）

代码 21.4.3：集群调度策略仿真对比（FIFO / Fair / DRF）

21.5 性能与可扩展性分析

21.6 关键要点

21.7 常见陷阱与注意事项

21.8 思考题（带答案）

Share on