20 · 批处理：可重跑的分组、连接和派生数据

目标：构造输入固定、输出确定且可整体发布的日报计算。

一次作业究竟处理哪一份世界

批处理面对有界输入，作业可以完成。如果输入在处理中改变，同一次计算就可能混入不同时刻的数据；如果输出不断追加到现有结果，重跑可能重复累加。可靠重跑通常需要固定输入版本、明确转换规则和独立输出，完成后再发布。

Map 把每条记录转成待分组键值，shuffle 把相同键送到同一处理方，reduce 合并该键的值。shuffle 是按键重新分布，不是随机打乱。DataFrame 和 SQL 可以让用户声明转换关系，执行器再安排扫描、连接、分组与数据移动。

小维度表可以复制到每个工作者本地做连接；大表连接则可能要求双方按连接键重分布。一个热门键会让某个 reducer 很忙，因此总资源足够也可能有长尾任务。内存不够时可用外部排序和磁盘中间文件，但需要计入 I/O 与恢复成本。

写完再替换日报

保存 batch.py。本例把不可变报名快照按城市分组，借临时文件演示完整输出发布；replace 只针对同一临时目录中的本地文件，不代表对象存储存在相同重命名语义。

import json
from pathlib import Path
from tempfile import TemporaryDirectory

registrations = [("r1", "a1"), ("r2", "a1"), ("r3", "a2")]
activity_city = {"a1": "north", "a2": "south"}

def build(rows):
    grouped = {}
    for registration_id, activity_id in rows:
        city = activity_city[activity_id]
        grouped.setdefault(city, []).append(registration_id)
    return {city: len(ids) for city, ids in sorted(grouped.items())}

expected = {"north": 2, "south": 1}
assert build(registrations) == expected
assert build(list(reversed(registrations))) == expected
with TemporaryDirectory() as work:
    target = Path(work) / "report.json"
    staging = Path(work) / "report.next.json"
    for _ in range(2):
        staging.write_text(json.dumps(build(registrations), sort_keys=True))
        staging.replace(target)
    observed = json.loads(target.read_text())
    assert observed == expected
    print(observed)

uv run --python 3.12 --no-project batch.py

预期输出：

{'north': 2, 'south': 1}

连接通过活动 ID 找城市；分组后计数；输出按键排序。这一转换没有读取当前时钟，也没有向外发通知，因而输入不变时可以重跑。如果活动城市会变，快照还应固定维度版本，否则旧报名可能被重算到新城市。

引导修改：① 增加 ('r4', 'a2')；② 更新预期为两个城市各 2；③ 连续重跑，结果应保持 2 和 2，不能变成 4 和 4。注意此处输入前提是每条报名唯一；重复原始事件不是第二份合法报名。

计算、存储与编排分开考虑

分布式文件系统和对象存储保存输入、中间或最终数据；计算框架决定如何执行算子；作业编排器分配资源和重试任务；工作流编排器决定哪些作业依赖哪些上游完成。某个任务失败后，必须防止它的部分输出和重试输出同时被计入。

Unix 的过滤、排序、计数管道也体现相同思想：输入输出接口简单，计算可以组合。规模变化后需要解决的是资源边界和失败发布，而不只是把函数换成集群 API。训练数据准备、离线索引构建和报表都能利用这一模式。

独立练习：某次作业只完成 north 就崩溃。读者应继续看到什么？如果数据放在对象存储，提出不依赖目录重命名的发布方式。

缺城市时检查维度表，而不是默默归为“其他”；重跑翻倍时检查是否对已发布结果继续追加。在 design.md 记录日报输入快照、城市时间口径和完整版本发布规则。

参考资料

• DDIA 第二版第 11 章：批处理、shuffle、连接、编排与派生数据发布。