多租户、配额与切换：从令牌桶到代际租约，以及 checkpoint 不是业务提交

U85 讲清调度器如何触发与收敛，U86 讲清引擎提交与 Notebook 内核的状态错位。这一页把两者放进同一个"任务 + 租户"的案例：多个租户共享一套调度与执行资源时，谁先被允许跑（配额）、谁能代表系统做决定（代际租约）、恢复时重放什么、不重放什么（checkpoint 与业务提交）。三件事各自都有真实的执行点与失败边界，混在一起讲会导致"用了 checkpoint 就以为业务恰好一次"这类致命误解。

可运行结论来自 labs/data-tenant-quota：真实 MySQL 8.4.7 上的令牌桶并发扣减、真实子进程 SIGKILL 后的代际租约接管、真实 Spark 4.0.1 Structured Streaming 的 checkpoint 重放与幂等写，共 4 组场景一次完整运行 exit 0。本页是 U87（#124）的 D-DATA 方向正文。

本页不重复：租户隔离与数据生命周期讲隔离层级与数据生命周期；分布式锁、分布式 ID 与租约讲持久任务状态机的纸面模型；DAG 调度与恢复讲调度器本体；执行引擎与 Notebook 内核讲提交/状态/结果三层错位。本页只补运行期的资源裁决、代际接管与checkpoint 与业务提交的边界。

术语：租户、配额、代际、checkpoint、业务提交

五个词在本页有固定含义：

术语	定义	不是什么
租户	共享平台上的一个资源与数据所有者	不是"用户账号"；一个租户可含多个用户
配额	对某租户在某维度上的使用上限（本实验是令牌数）	不是限流中间件；本实验在提交前裁决
代际（generation）	同一名字的租约被接管时递增的整数	不是时间戳；时间戳会回拨，代际单调
checkpoint	引擎记录的"我处理到哪"（偏移量 + 状态）	不是业务效果；checkpoint 成功不代表外部写入已发生
业务提交	业务效果真正落到权威存储（本实验是 MySQL 业务表）	不是引擎的 commit log；两者可单独成功

整页的核心命题：

配额决定"能不能跑"，代际决定"谁有资格决定"，checkpoint 决定"从哪继续"——三者都不能代替"业务效果已经发生且只发生一次"。

版本、工程与观察入口

固定版本：MySQL 8.4.7（本机 DBngin）、Spark 4.0.1（复用 U86 缓存的官方发行包，SHA-512 校验）、Python 3.9 + pymysql（worker 子进程）。运行时 JDK 21.0.11、macOS 27.2 arm64。Spark 是 local[2]：没有集群级资源队列，所以本页的"配额"是应用层裁决，不是 YARN/K8s 队列——这个边界在后文每节都会重申。

公开入口：

实验代码sh

python3 labs/data-tenant-quota/run.py

每次运行创建独有库 study_u87_<runid>、独有临时目录与 Spark checkpoint 目录；结束停进程、DROP 数据库、清理孤儿库；清理不完整使退出码非 0。完整场景与清理结果见运行证据，工程边界见实验说明。

配额：提交前的原子裁决

为什么配额必须在提交前

任务提交后再"发现超配额"只能取消，而取消不保证已产生的副作用消失（U85 的取消边界）。所以配额的执行点应当在任务进入队列/执行器之前：原子的"检查并扣减"，失败即拒绝，不进入执行。

本实验用 MySQL 单行实现令牌桶——tenant_quota(tenant, tokens, updated_at)，扣减条件是"还有令牌"：

sql

-- 教学代码：labs/data-tenant-quota 的配额扣减（MySQL 8.4 实跑）
UPDATE tenant_quota
SET tokens = tokens - 1, updated_at = NOW()
WHERE tenant = :tenant AND tokens > 0;
-- rowcount = 1 表示扣减成功；0 表示配额不足，任务应被拒绝且不进入执行

这条 UPDATE ... WHERE tokens > 0 本身就是原子裁决：并发的两个请求，只有一个能让 rowcount=1。实测 6 个线程、两个租户各请求 3 次、初始各 5 个令牌：

text

U87OBS quotaRequests tenantA=3 tenantB=3 total=6
U87 quotaRemaining tenantA=2 tenantB=2 requestedEach=3

每个租户恰好成功 3 次、剩余 2——没有超扣，也没有互相干扰。这验证了两件事：单行条件更新在 InnoDB 行锁下是原子的；租户隔离只要落在不同的行上，天然互不影响。

配额的维度与"超限之后"

本实验只做了一个维度（次数令牌），真实平台通常同时有并发、速率、存储与昂贵查询四类（见租户隔离与数据生命周期的公平分摊一节）。无论几个维度，执行点都有同一个要求：裁决与扣减放在同一次原子操作里，否则并发下会"检查通过、扣减失败"或"多个请求同时通过"。

超限后的行为是产品决策，本实验验证了两种：

text

U87 quotaGate blockedStatus=REJECTED tokensAtBlock=0
U87 quotaGate recoveredStatus=ACCEPTED tokensAfter=2

配额为 0 时任务状态写 REJECTED 且不产生任何业务行；恢复配额（3 个令牌）后同一任务 ACCEPTED、令牌剩 2。拒绝是可观测、可恢复的状态，而不是静默丢弃——这一点与 U85 的 DISCARD_LATER 同构：被拒绝的请求没有执行痕迹，但必须在登记层留痕，否则运营看不到"谁被限了"。

两种实现方案的取舍

方案	原子性来源	成本	适用
MySQL 行锁（本实验）	`UPDATE ... WHERE` 的行锁	每次提交一次 DB 往返	已有关系库、提交频率不高的平台
Redis 原子脚本 / 令牌桶服务	Lua 脚本或服务端 CAS	更快，但引入第二个存储与一致性边界	高 QPS、需要毫秒级裁决

选 MySQL 的代价要说清：DB 抖动会直接影响"能否提交"，配额存储的可用性变成提交路径的可用性。选 Redis 则要处理"配额已扣但任务未提交成功"的补偿。两者都没有免费午餐。

代际租约：接管必须递增，旧代际必须被拒

"谁有资格做决定"在分布式系统里通常用租约表达。U85 的 xxl-job 用数据库行锁做调度互斥，但那是"互斥"，不是"所有权"：它保证同一时刻只有一个进程在扫描，却不记录"现在是第几代"。当需要旧持有者复活后不能继续写时，就需要代际。

本实验的租约行：lease_state(name, owner, generation, expires_at)。获取条件是"无主或已过期"，且成功后 generation + 1：

sql

-- 教学代码：获取或接管租约（条件更新 + 代际递增）
UPDATE lease_state
SET owner = :new_owner, generation = generation + 1,
    expires_at = DATE_ADD(NOW(), INTERVAL :lease_seconds SECOND)
WHERE name = :name AND (owner IS NULL OR expires_at < NOW());
-- 接管成功后 generation 必然比旧代际大 1；旧代际的后续写入应带
-- AND generation = :old_generation 条件，从而影响 0 行

-- 旧代际的迟到写入（必须被拒）
UPDATE lease_state SET expires_at = DATE_ADD(NOW(), INTERVAL 60 SECOND)
WHERE name = :name AND owner = :old_owner AND generation = :old_generation;

完整轨迹（真实子进程，真实 SIGKILL；接管者与旧持有者同名——这是让 owner 失效、只剩代际可用的关键设计）：

text

U87 leaseAcquire holder=worker-a generation=1
U87 leaseHolderKilled signal=SIGKILL
U87OBS leaseTakeover owner=worker-a affected=1 generation=2
U87OBS leaseOwnerOnlyMatches owner=worker-a count=1
U87OBS leaseStaleWrite owner=worker-a generation=1 affected=0
U87 leaseFinal owner=worker-a generation=2

五个观察对应五个事实：

worker-a 以 gen1 成为持有者；
worker-a 被 SIGKILL（模拟主节点崩溃，不是优雅退出）；
过期后 同名重启的 worker-a 接管，affected=1、generation 变成 2；
只按 owner 匹配仍然命中 1 行（leaseOwnerOnlyMatches count=1）——owner 无法区分新旧进程；
旧进程"复活"用 gen1 做条件更新——affected=0，被拒。

第 5 步是代际的全部意义。在本实验的 schema 里，接管后 owner 保持同名，唯一的区分就是 generation；如果写路径只按 owner 判断，旧进程会把自己当成合法持有者，加上 generation=1 后写路径自动失效。这与分布式锁、分布式 ID 与租约的 claim_version 是同一思想，只是本实验用真实 MySQL 跑通了完整往返。

接管不等于旧尝试没有副作用

实验中 worker-a 被 SIGKILL 前没有做任何业务写，所以接管是干净的。现实中旧持有者可能在"已经发出写请求、还没记录完成"时被杀——这时接管者会重做，而旧请求可能已生效。代际只保证"旧代际的后续写入被拒"，不保证"旧代际已发出的请求没有落地"。要对冲这一点，业务侧必须有稳定幂等键与查单（与 U85 的"租约到期只允许接管、不证明旧尝试未产生副作用"完全一致）。

代际 vs 时间戳 vs 行锁

机制	能回答	不能回答
时间戳 `expires_at`	租约是否过期	谁是新持有者、旧持有者是否复活
行锁 `FOR UPDATE`（U85 调度）	此刻谁在扫描	谁"应该"在扫描、旧会话是否复活
代际 `generation`	谁有资格写、旧代际是否失效	旧代际已发出的副作用

三者叠加才是完整的"所有权"：过期判断 + 互斥 + 代际单调递增。

checkpoint 与业务提交：两条独立的事实

两条记录分别是什么

Spark Structured Streaming 的 checkpoint 目录里，引擎维护 offset log 与 commit log：offset log 记录"这批要处理哪些数据"，commit log 记录"这批已处理完"。每批的处理顺序在 MicroBatchExecution 里写得很清楚——先执行 sink：

原码scala

    val nextBatch =
      new Dataset(execCtx.executionPlan, ExpressionEncoder(execCtx.executionPlan.analyzed.schema))

    val batchSinkProgress: Option[StreamWriterCommitProgress] =
      execCtx.reportTimeTaken("addBatch") {
      SQLExecution.withNewExecutionId(execCtx.executionPlan) {
        sink match {
          case s: Sink =>
            s.addBatch(execCtx.batchId, nextBatch)
            // DSv2 write node has a mechanism to invalidate DSv2 relation, but there is no
            // corresponding one for DSv1. Given we have an information of catalog table for sink,
            // we can refresh the catalog table once the write has succeeded.
            plan.catalogTable.foreach { tbl =>
              sparkSession.catalog.refreshTable(tbl.identifier.quotedString)
            }
          case _: SupportsWrite =>
            // This doesn't accumulate any data - it just forces execution of the microbatch writer.
            nextBatch.collect()
        }
        execCtx.executionPlan.executedPlan match {

sink 成功之后，才把本批写进 commit log（commitLog.add）：

原码scala

    if (StatefulOperatorStateInfo.enableStateStoreCheckpointIds(
      sparkSessionForStream.sessionState.conf)) {
      updateStateStoreCkptId(execCtx, latestExecPlan)
    }
    execCtx.reportTimeTaken("commitOffsets") {
      val stateStoreCkptId = if (StatefulOperatorStateInfo.enableStateStoreCheckpointIds(
        sparkSessionForStream.sessionState.conf)) {
        Some(currentStateStoreCkptId.toMap)
      } else {
        None
      }
      if (!commitLog.add(execCtx.batchId,
        CommitMetadata(watermarkTracker.currentWatermark, stateStoreCkptId))) {
        throw QueryExecutionErrors.concurrentStreamLogUpdate(execCtx.batchId)
      }
    }
    committedOffsets ++= execCtx.endOffsets
  }

由此得到一条重要事实：checkpoint 的推进在 sink 之后。如果 sink 抛异常，这一批不会进 commit log，重启后会重新读取同一批数据。这正是本实验要验证的行为——也是"checkpoint 成功 ≠ 业务提交成功"的另一面：checkpoint 只在 sink 成功后才推进，而 sink 成功不代表外部业务表真的写入（sink 内部可能先写外部、后本地提交，或反之）。

四批真实运行（前三批 + 一次篡改）

实验用 foreachBatch 手动控制业务写，故意在第二批注入失败；最后再删掉 commit log 里 batch 1 的提交记录，强制引擎真正重复投递：

text

U87 checkpointFirstRun exit=0 businessRows=5
U87 checkpointFailedBusiness exit=1 businessRows=5
U87 checkpointReplay exit=0 businessRows=8
U87 checkpointReplayBatches epoch1Reprocessed=True
U87 checkpointReplayAgain exit=1 businessRows=8 epoch1Reprocessed=True tamperGuardFired=True

逐条解释：

第一批：读到 5 行，业务表写入 5 行，exit 0。
第二批：读到 3 行新数据，业务写抛异常，driver 非零退出；业务表仍是 5 行——这一批的业务没有提交，checkpoint 也没有推进到这批之后。
重放：重启后引擎从 commit log 的上一批继续，重新读到那 3 行（epoch1Reprocessed=True），写入成功，业务表 8 行，exit 0。
篡改 commit log 后重放：删除 commits/1 后重启，引擎真的重新处理了 batch 1（epoch1Reprocessed=True，日志重现 U87BATCH epoch=1）；业务表仍是 8 行——唯一键 + INSERT IGNORE 让同一批重复投递没有产生重复行。随后 Spark 4 在提交时检测到 commit log 与已处理状态不一致，以 STREAM_FAILED（Multiple streaming queries are concurrently using .../commits）拒绝继续（tamperGuardFired=True，退出码 1）。

第 3 步证明"重放的是 sink 失败的那一批，不是全部历史"；第 4 步同时给出两个结论：业务侧幂等确实把重复投递收敛为"效果一次"（行数仍 8），而引擎自身对被篡改的 checkpoint 有完整性保护——它会把这类情况当作错误终止，而不是悄悄继续。生产上删除 checkpoint 或提交记录来做"重跑"是危险操作，本实验展示了它的实际后果。

关于幂等的边界：如果业务表没有唯一键，第 4 步的重复投递就会插入重复行——"checkpoint 推进了"与"业务恰好一次"之间的差距会立刻暴露成账实不符。checkpoint 提供 at-least-once 的重放，恰好一次的效果必须由业务侧收敛。

常见的三个误解

误解	实际
"checkpoint 成功 = 业务成功"	checkpoint 在 sink 成功后推进，但 sink 自身可能只写了一半（例如先写外部存储、再写本地元数据）
"checkpoint 能保证恰好一次"	引擎保证 at-least-once 重放；恰好一次取决于 sink 的幂等性
"重放会从头开始"	只重放 commit log 之后未提交的批次；已提交批次不会重跑（除非手动删除 checkpoint）

组合：配额、代际与 checkpoint 同时出现

三者在真实平台上会同时出现，责任必须分清：

事件	由谁裁决	观察点
提交被拒（超配额）	配额执行点（提交前）	任务登记为 REJECTED，无业务行
调度/执行权切换	代际租约（持锁者 + generation）	lease_state 的 owner/generation
崩溃后从哪继续	checkpoint（commit log 偏移）	checkpoint 目录与重放的 epoch
业务效果是否恰好一次	业务幂等（唯一键/幂等键）	业务表的行数与唯一键冲突计数

错误的责任叠加会产生真实事故：把"配额拒绝"实现成"提交后取消"，取消不保证已产生副作用消失；把"代际接管"实现成"按 owner 判断"，复活的旧进程会覆盖新持有者的状态；把"checkpoint 成功"当成"业务成功"，对账时会发现数据缺失。本实验的 4 组场景分别验证了正确实现的边界，也给出了每类错误的可观测信号。

观测与排障

一次"任务没跑/跑重了/结果丢了"的排障顺序，把三组机制串起来：

配额层：查任务登记状态。REJECTED 说明从未进入执行，不需要查 checkpoint；ACCEPTED 才往下走。
租约层：查 lease_state 的 owner 与 generation。若任务由持有者 A 提交、但完成记录来自 B，检查是否发生了接管，以及旧代际是否绕过条件更新。
checkpoint 层：查 checkpoint 的 commit log 最新批次与业务表行数。commit log 领先业务表 → sink 部分失败；业务表领先 commit log → 说明有绕过 checkpoint 的手工写入。
幂等层：查业务表唯一键冲突计数。重放次数增加但冲突计数也增加，说明幂等生效；行数直接增加则说明没有唯一键。

容量与取舍

配额维度的选择：只做次数令牌最便宜，但挡不住"单任务很贵"（一次扫描全表）；真实平台需要并发 + 昂贵查询两类配额，代价是每次提交多几次原子操作。本实验只做了次数维度，其他维度是设计建议，未运行。

租约的过期时间：太短会频繁误判（网络抖动导致接管）、太长则故障恢复慢。本实验用 3 秒（为了在秒级观察接管），生产通常用 30 秒到几分钟，并按心跳周期设置。时钟是所有租约方案的隐含依赖：时钟回拨会让 NOW() 变小、条件 expires_at < NOW() 更难成立，接管被延迟；向前跳变则相反，会让尚未真正过期的租约被提前接管。代际能防"旧写入"，防不了"时间判断错误"。

checkpoint 间隔与业务批大小：trigger(availableNow=True) 让每次提交处理所有可用数据，适合批处理；流式常驻则按处理时间/数据处理量触发。间隔越大，重放时单批越大、恢复越慢。

演进方向：租户变多 → 配额需要独立服务（Redis/令牌桶服务），把裁决从业务库剥离；执行变重 → 需要集群级配额（YARN 队列、K8s ResourceQuota），应用层配额退化为"提交前的第一道闸"；切换要求变高 → 代际租约要配合 fencing token 传给下游（例如存储层校验 token 才接受写入），否则旧持有者仍可能写入不校验代际的存储。

实验读法

4 组场景与观察行（同一 exit 0 运行）：

U87OBS quotaRequests tenantA=3 tenantB=3 total=6、U87 quotaRemaining tenantA=2 tenantB=2：令牌桶原子扣减与租户隔离。
U87 leaseAcquire generation=1 → leaseTakeover owner=worker-a generation=2 → leaseOwnerOnlyMatches count=1 → leaseStaleWrite affected=0：同名接管、owner 无法区分、代际拒绝旧写入。
U87 checkpointFirstRun exit=0 businessRows=5 → checkpointFailedBusiness exit=1 businessRows=5 → checkpointReplay businessRows=8 epoch1Reprocessed=True → checkpointReplayAgain exit=1 businessRows=8 tamperGuardFired=True：失败批重放、篡改 commit log 后真实重复投递、幂等收敛与引擎完整性保护。
U87 quotaGate blockedStatus=REJECTED businessRowsBlocked=0 → recoveredStatus=ACCEPTED businessRowsAccepted=1：配额拒绝无业务行、恢复后正常写入。

开发中真实踩到的坑都留在缓存 runtime/ 的失败运行里：MySQL 批处理输出行尾空字段被 strip() 吃掉（改用哨兵值 -）、Python % 格式化与 SQL 的 %s 冲突（改用 f-string 生成 worker 脚本）——没有一个是"命令写好就算过"。

条件变化与未运行边界

未运行：集群级配额（YARN 队列、K8s ResourceQuota）、Redis 令牌桶、跨机租约（多台物理机时钟偏差）、Flink 的 checkpoint 与 savepoint、fencing token 下沉到存储层。
受控构造：租约过期用 3 秒（生产通常更长）；checkpoint 失败用 foreachBatch 主动抛异常；SIGKILL 是真实进程信号，未覆盖网络分区下的"假死"（进程还在但心跳丢失）。
local[2] 局限：Spark 没有集群资源协商，本页的"配额"是应用层，不能外推到集群调度器。
幂等的边界：本实验用唯一键 + INSERT IGNORE 收敛重复；对"更新类"业务，幂等需要版本号或状态机，不是唯一键能覆盖的。
checkpoint 重放保证的是同一批数据被重新处理，不保证下游副作用可撤销；如果 sink 已经对外部系统产生了不可逆效果（付款、发消息），重放前必须先查单。

闭卷验收

本页 20 道闭卷题收进了下方自测面板：先回忆，再展开参考答案与评分点，最后给三档自评。

来源与版本

来源	版本	用途
Spark `MicroBatchExecution.scala`	v4.0.1	sink 执行与 commit log 推进顺序（真实运行）
Spark 4.0.1 官方发行包	4.0.1	Structured Streaming 真实运行
MySQL 8.4 参考手册：InnoDB 锁	8.4	行锁与条件更新语义
分布式锁、分布式 ID 与租约	课程正文	claim/lease 模型与旧尝试边界
租户隔离与数据生命周期	课程正文	隔离层级与配额维度

访问与运行日期：2026-10-28。源码片段为固定 tag 的连续行段，整文件 SHA-256 随每段元数据登记；运行版本以证据 JSON 为准。产品文档、源码与实验是三类证据，不能互相替代；集群级配额与 Flink checkpoint 未运行。

多租户、配额与切换：从令牌桶到代际租约，以及 checkpoint 不是业务提交 ​

术语：租户、配额、代际、checkpoint、业务提交 ​

版本、工程与观察入口 ​

配额：提交前的原子裁决 ​

为什么配额必须在提交前 ​

配额的维度与"超限之后" ​

两种实现方案的取舍 ​

代际租约：接管必须递增，旧代际必须被拒 ​

接管不等于旧尝试没有副作用 ​

代际 vs 时间戳 vs 行锁 ​

checkpoint 与业务提交：两条独立的事实 ​

两条记录分别是什么 ​

四批真实运行（前三批 + 一次篡改） ​

常见的三个误解 ​

组合：配额、代际与 checkpoint 同时出现 ​

观测与排障 ​

容量与取舍 ​

实验读法 ​

条件变化与未运行边界 ​

闭卷验收 ​

来源与版本 ​

多租户、配额与切换：从令牌桶到代际租约，以及 checkpoint 不是业务提交

术语：租户、配额、代际、checkpoint、业务提交

版本、工程与观察入口

配额：提交前的原子裁决

为什么配额必须在提交前

配额的维度与"超限之后"

两种实现方案的取舍

代际租约：接管必须递增，旧代际必须被拒

接管不等于旧尝试没有副作用

代际 vs 时间戳 vs 行锁

checkpoint 与业务提交：两条独立的事实

两条记录分别是什么

四批真实运行（前三批 + 一次篡改）

常见的三个误解

组合：配额、代际与 checkpoint 同时出现

观测与排障

容量与取舍

实验读法

条件变化与未运行边界

闭卷验收

来源与版本