请求（request）和 查询（query），以及相应的 响应（response）或 结果（result）。许多现代数据系统都默认采用这种处理方式：你请求某样东西，或者发出一条指令，系统便尽量快速地给出答案。
浏览器请求网页、服务调用远程 API，以及数据库、缓存、搜索索引等许多系统，都是这样工作的。我们称它们为 在线系统（online system）。这类系统通常以响应时间作为主要性能指标，而且往往需要具备容错能力，才能保证高可用性。
有些计算规模太大，或者需要处理的数据太多，无法放在一次交互式请求中完成。例如，你可能需要训练 AI 模型，把大量数据从一种形式转换成另一种形式，或者在非常大的数据集上进行分析。我们把这类任务称为 批处理（batch processing）作业，相应的系统有时也称为 离线系统（offline system）。
批处理作业读取一组输入数据（只读），并产生一组输出数据（每次运行都从头生成）。它通常不会像读写事务那样修改现有数据。因此，输出是由输入 衍生（derived）而来的。 如果对输出不满意，只需把它删除，调整作业逻辑，再运行一次。
一项批处理作业可能要运行很长时间：几分钟、几小时，甚至几天。作业也可能按固定周期调度运行，例如每天一次。其主要性能指标通常是吞吐量，即单位时间内能够处理多少数据。有些批处理系统遇到故障时只会中止并重新启动整个作业；另一些则具备容错能力，即使某些节点崩溃，作业也能顺利完成。
现代批处理深受 MapReduce 影响，Google 2004 发表算法，随后 Hadoop、CouchDB、MongoDB 等多个开源数据系统都实现了它。MapReduce 是一种相当底层的编程模型，如今基本过时了，Google 都不用了。
如今，批处理更多由 Spark、Flink 之类的框架或数据仓库查询引擎完成。与 MapReduce 一样，这些系统高度依赖分片和并行执行，但它们的缓存和执行策略精巧得多。
云计算已经十分普及，批处理的存储层也从 HDFS、Gluster、CephFS 等分布式文件系统 DFS 转向 S3 之类的对象存储。BigQuery、Snowflake 等可伸缩的云数据仓库，则进一步模糊了数据仓库与批处理系统之间的界限。
假设有一台 Web 服务器，每处理一个请求，它都会在日志文件末尾追加一行。以 nginx 默认的访问日志格式为例，其中一行可能如下所示：
216.58.210.78 - - [27/Jun/2025:17:55:11 +0000] "GET /css/typography.css HTTP/1.1" 200 3377 "https://martin.kleppmann.com/" "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/137.0.0.0 Safari/537.36"解析日志是许多现代科技公司的关键工作，从广告数据管道到支付处理无所不在。
许多工具都能读取这些日志文件，生成漂亮的网站流量报告。来用基本的 Unix 工具自行构建一个。假设你想找出网站上最受欢迎的五个页面，可以在 Unix shell 中执行下面的命令：
cat /var/log/nginx/access.log | #1
awk '{print $7}' | #2
sort | #3
uniq -c | #4
sort -r -n | #5
head -n 5 #6Unix 工具，非常值得学习，仅用 awk sed
grep sort uniq xargs
就能完成许多数据分析，而且性能还不错。
除了使用 Unix 命令链，可以写一个简单的程序来完成同样的工作。
from collections import defaultdict
counts = defaultdict(int) #1
with open('/var/log/nginx/access.log', 'r') as file:
for line in file:
url = line.split()[6] #2
counts[url] += 1 #3
top5 = sorted(((count, url) for url, count in counts.items()), reverse=True)[:5] #4
for count, url in top5: #5
print(f"{count} {url}")Python 脚本在内存中维护一个 URL 散列表，把每个 URL 映射到它出现的次数。Unix 管道没有这样的散列表，而是依靠排序 URL 列表；同一个 URL 出现多次时，它只是在列表中重复多次。
如果作业的工作集大于可用内存，排序方法就有一个优势：它可以高效利用磁盘。先在内存中对数据块排序，并把它们作为段文件写入磁盘，再将多个有序段合并成一个更大的有序文件。归并排序采用顺序访问模式，在磁盘上表现很好。
GNU Coreutils（Linux）中的 sort 工具会把无法装入内存的数据自动溢写到磁盘，还会自动利用多个 CPU 核并行排序。这意味着前面的简单 Unix 命令链可以轻松扩展到大型数据集，而不会耗尽内存。此时的瓶颈很可能是从磁盘读取输入文件的速度。
Unix 工具的局限在于，它们只能在单台机器上运行。如果数据集大到无法装入单机内存或本地磁盘，就会遇到问题，这正是分布式批处理框架的用武之地。
有时写各种程序，搞各种算法，可能没有大力出奇迹方式高效，不如多买点内存把要处理的数据全部加载到内存然后在内存上作业 可能比搞什么分布式系统还要好。
运行前面 Unix 工具示例的机器，需要由几个组件协同处理日志数据：
分布式数据处理框架中也存在同样的组件。事实上，你可以把 分布式处理框架（distributed processing framework）看作一种分布式操作系统：它们拥有文件系统和作业调度器，程序则通过文件系统或其他通信通道相互发送数据。
操作系统提供的文件系统由若干层组成：
分布式文件系统（distributed filesystem）的工作方式与此非常相似。文件同样被拆分成块，只不过这些块分布在许多机器上。分布式文件系统的块通常比本地文件系统大得多：HDFS（Hadoop 分布式文件系统）的默认块大小为 128 MB，JuiceFS 和许多对象存储使用 4 MB 的块，而 ext4 的块只有 4,096 字节。块越大，需要跟踪的元数据就越少；对于 PB 级数据集，这种差异十分显著。相对于读取数据块所需的时间，大块也能降低寻道开销所占的比例。
大多数物理存储设备无法写入不完整的数据块，因此即使数据没有填满一个块，操作系统也必须为写入使用整个块。分布式文件系统的块更大，而且通常构建在操作系统文件系统之上，所以没有这种要求。例如，一个 900 MB 的文件采用 128 MB 的分块时，会由 7 个占用 128 MB 的块和 1 个占用 4 MB 的块组成。
读取分布式文件系统中的块，需要向集群中存放该块的机器发出网络请求。每台机器都运行一个守护进程，并公开一套 API，让远程进程能够读写以文件形式存储在其本地文件系统中的数据块。HDFS 把这些守护进程称为 DataNode，GlusterFS 则称为 glusterfsd。本书统一把它们称为 数据节点（data node）。
分布式文件系统还实现了分布式版本的页缓存。由于数据块以文件形式存放在数据节点上，读写操作会经过每个数据节点的操作系统，其中就包含内存页缓存。因此，经常读取的数据块会被缓存在数据节点的内存中。有些分布式文件系统还实现了更多缓存层，例如 JuiceFS 提供的客户端缓存和本地磁盘缓存。
ext4 和 XFS 等文件系统会跟踪空闲空间、文件块位置、目录结构和权限设置等存储元数据。分布式文件系统同样需要记录文件分散在哪些机器上、具有什么权限等信息。Hadoop 通过 NameNode 服务维护集群元数据；DeepSeek 的 3FS 则使用元数据服务，并把数据持久化到 FoundationDB 之类的键值存储中。
文件系统层之上是 VFS；在批处理中，与之最接近的是分布式文件系统的协议。分布式文件系统必须公开某种协议或接口，让批处理系统能够读写文件。它充当一套可插拔接口：只要实现该协议，任何分布式文件系统都可以接入。例如，MinIO、Cloudflare R2、Tigris 和 Backblaze B2 等许多存储系统都采用了 Amazon S3 API。支持 S3 的批处理系统，也就能够使用其中任何一种存储。
有些分布式文件系统提供与 POSIX 兼容的接口，在操作系统的 VFS 看来，它们与其他文件系统并无不同。这类系统通常通过用户空间文件系统（FUSE）或网络文件系统（NFS）协议接入 VFS。NFS 或许是最著名的分布式文件系统协议。它最初的用途，是让多个客户端读写单台服务器上的数据。近来，AWS Elastic File System（EFS）和 Archil 等文件系统提供了可伸缩得多、但仍与 NFS 兼容的分布式实现。NFS 客户端依旧只连接一个端点，不过这些系统会在内部与分布式元数据服务和数据节点通信，完成数据读写。
许多分布式文件系统构建在普通商用硬件上，这种硬盘比较便宜，故障率高于企业级硬件。为了容忍机器和磁盘故障，文件块会被复制到多台机器上。这样也便于调度器更均匀地分配工作负载，因为任务可以在任何存有其输入数据副本的节点上运行。在多台机器上保存若干份完整副本，可以采用 Reed-Solomon 码等 纠删码（erasure coding）， 以低于完整复制的存储开销恢复丢失的数据。技术和 RAID 很相似，后者在连接到同一台机器的多个磁盘之间提供冗余。区别在于，分布式文件系统通过普通的数据中心网络访问和复制文件，不需要特殊硬件。
Amazon S3、Google Cloud Storage、Azure Blob Storage 和 OpenStack Swift 等 对象存储（object storage）服务，已经成为批处理作业中分布式文件系统的常用替代方案。用户空间文件系统（FUSE）驱动程序可以让用户把 S3 之类的对象存储当作文件系统使用。JuiceFS 和 Ceph 等分布式文件系统实现也同时提供对象存储与文件系统 API。不过，两类系统的 API、性能和一致性保证可能大相径庭。即使某个系统看似实现了所需的 API，采用之前也必须仔细确认它的实际行为符合预期。
对象存储中的每个对象都有一个 URL，例如
s3://my-photo-bucket/2025/04/01/birthday.png，URL
的主机部分（my-photo-bucket）表示存放对象的存储桶（bucket），后面的部分则是对象的
键（本例中为 /2025/04/01/birthday.png
）。存储桶的名称在全局范围内唯一，而每个对象的键在所属存储桶内必须唯一。
对象通过 get 调用读取，通过 put 调用写入。与文件系统中的文件不同，对象写入后是不可变的；若要更新，只能像键值存储一样，通过 put 完整重写整个对象。Azure Blob Storage 和 S3 Express One Zone 支持追加写入，但大多数对象存储都不支持。对象存储也没有 fopen、fseek 之类的文件句柄 API。
对象看起来似乎按目录组织，其实对象存储根本没有目录的概念，路径结构只是一种约定，斜杠本身也是对象键的一部分。这种约定允许你按特定前缀请求对象列表，实现类似目录列表的效果。不过，按前缀列出对象与文件系统列出目录有两点区别：
s3://my-photo-bucket/2025/04/01 下的所有对象，那么对
s3://my-photo-bucket/2025/04 调用 list 时，01
就不会再出现。常见的做法是用一个零字节对象来表示空目录，例如创建空对象
s3://my-photo-bucket/2025/04/01，这样即使所有子对象都已删除，它仍会保留下来。分布式文件系统通常支持硬链接、符号链接、文件锁和原子重命名等常见文件操作，对象存储则没有这些功能。它们一般不支持链接和锁，重命名也不是原子的，而是先把对象复制到新键，再删除旧对象。如果想重命名一个“目录”，就必须逐一重命名其中的每个对象，因为目录名本身是对象键的一部分。
键值存储针对小值（通常只有几 KB）以及频繁、低延迟的读写进行了优化。相比之下 分布式文件系统和对象存储通常针对大对象（数 MB 至数 GB）和频率较低、规模较大的读取进行了优化。不过最近，对象存储也开始支持更加频繁、规模更小的读写。例如，S3 Express One Zone 如今可以达到个位数毫秒延迟，其定价模型也更接近键值存储。
分布式文件系统与对象存储还有一个区别：
操作系统的类比同样适用于作业编排。执行 Unix 批处理作业时，总要有某个组件真正运行 awk、sort、uniq 和 head 这些进程。它必须把一个进程的输出传到另一个进程的输入，为每个进程分配内存，在 CPU 上公平地调度并执行各进程的指令，实施内存和 I/O 边界，等等。在单台机器上，这些工作由操作系统内核负责；在分布式环境中，它们则由作业编排器承担。
批处理框架会向编排器的调度器发送运行作业的请求。启动作业的请求包含如下元数据：
Kubernetes 和 Hadoop YARN（Yet Another Resource Negotiator）等编排器会结合这些信息与集群元数据，通过下列组件执行作业：
任务执行器（Task Executor）:
集群中的每个节点都运行着执行器守护进程，例如 YARN 的 NodeManager 或 Kubernetes 的 kubelet。执行器负责运行作业任务，发送心跳来表明自己仍然存活，并跟踪节点上的任务状态与资源分配。执行器收到启动任务的请求后，会取得作业的可执行代码，并运行命令来启动任务。然后，它会监视进程，直到进程结束或失败，再相应更新任务状态元数据。
许多执行器还会与操作系统配合，同时提供安全隔离与性能隔离。例如，YARN 和 Kubernetes 都使用 Linux cgroups。这样既能防止任务访问无权访问的数据，也能防止任务过度使用资源、影响同一节点上其他任务的性能。
资源管理器（Resource Manager）:
编排器的资源管理器保存每个节点的元数据，包括可用硬件（CPU、GPU、内存和磁盘等）、任务状态、网络位置、节点状态以及其他相关信息。因此，资源管理器能够提供集群当前状态的全局视图。资源管理器的中心化特性可能同时形成可伸缩性和可用性瓶颈。YARN 使用 ZooKeeper、Kubernetes 使用 etcd 来存储集群状态。
调度器（Scheduler）:
编排器通常有一个中心化的调度子系统，负责接收启动、停止作业或查询作业状态的请求。例如，调度器可能收到一项请求：使用特定的 Docker 镜像，在配有某种 GPU 的节点上启动一项包含 10 个任务的作业。调度器根据请求中的信息和资源管理器保存的状态，决定把哪些任务放在哪些节点上运行。随后，它会把分配结果通知任务执行器，由执行器开始运行任务。
有些调度决策需要由应用专用的调度器来完成，以便考虑特定需求，例如在查询量达到某个阈值时自动扩展只读副本。中心调度器与应用专用调度器共同决定任务的最佳执行方式。YARN 把这种子调度器称为 ApplicationMaster，Kubernetes 则称为 operator。
调度器在作业编排中扮演着格外棘手的角色：面对需求相互竞争的作业，它必须找出分配集群有限资源的最佳方式。从根本上说，调度决策需要在公平与效率之间取得平衡。
设想一个由五个节点组成的小型集群，总共有 160 个 CPU 核可用。集群调度器收到了两项作业请求，每项都希望使用 100 个核来完成工作。怎样调度才最好？
这个例子很简单，却已经暴露出许多艰难的权衡，以成组调度为例：如果调度器不断预留 CPU 核，直到 100 个核能够同时使用，那么一些节点就会闲置，集群的资源利用率也会下降；如果其他作业也试图预留 CPU 核，甚至还可能发生死锁。
另一方面，如果调度器只是等待 100 个核空闲，其他作业又可能在此期间抢走这些核。集群或许会在很长时间内都凑不出 100 个可用核，从而导致 饥饿（starvation）。调度器还可以 抢占（preempt）第一项作业的部分任务，终止它们来为第二项作业腾出空间。不过，抢占任务同样会降低集群效率，因为被终止的任务稍后需要重新启动、重新执行。
现在再设想一下，调度器必须为数百乃至数百万项这样的作业请求作出分配决策。要找到最优解似乎根本不可行。事实上，这个问题是 NP-hard 的；也就是说，除了规模最小的例子之外，计算最优解所需的时间长得令人无法接受。
因此，实际的调度器会采用启发式方法，作出虽非最优、但还算合理的决策。常见算法包括先进先出（FIFO）、主导资源公平（DRF）、优先级队列、基于容量或配额的调度，以及各种装箱算法。
开头的 Unix 工具示例由若干命令串联而成。分布式批处理也经常采用相同的模式：一项作业的输出需要成为另一项或多项作业的输入，而一项作业又可能有多个输入，分别由其他作业产生。这种作业结构称为 工作流（workflow），也称为作业的 有向无环图（directed acyclic graph，DAG）。
采用多项作业组成的工作流可能有几个原因：
在 Unix 工具示例中，连接一项命令输出与另一项命令输入的管道只使用一个很小的内存缓冲区，并不会把数据写入文件。如果缓冲区已满，生产数据的进程就必须等待，直到消费进程从缓冲区中读走一部分数据，才能继续输出，这是一种 背压 。Spark、Flink 等批处理执行引擎支持类似的模型，可以把一项任务的输出直接传给另一项任务；如果两项任务运行在不同机器上，数据就通过网络传输。
不过在工作流中，更常见的做法是让一项作业把输出写入分布式文件系统或对象存储，再由下一项作业从那里读取。这样可以使作业彼此解耦，在不同时间运行。如果一项作业有多个输入，工作流调度器通常要等到产生这些输入的所有作业都成功完成，才会运行消费这些输入的作业。
YARN ResourceManager 等编排框架中的调度器，以及 Spark 的内置调度器，都不会管理完整的工作流；它们只按单项作业进行调度。为了处理多次作业执行之间的依赖，人们开发了 Airflow、Dagster 和 Prefect 等工作流调度器。维护大量批处理作业时，这些调度器提供了非常有用的管理功能。许多数据管道的工作流通常包含 50 至 100 项作业；在大型组织中，还可能有许多团队运行不同的作业或工作流，跨越多个系统读取彼此的输出。管理这样复杂的数据流，离不开相应的工具支持。
批处理作业往往会运行很长时间。一项包含许多并行任务、长时间运行的作业，很可能会在途中遇到至少一次任务失败。 这可能由许多原因造成，其中包括硬件故障（普通商用硬件上尤其常见）和网络中断。
任务无法完成的另一个原因，是调度器可能有意抢占（终止）它。当系统设置多个优先级时，抢占尤其有用：低优先级任务运行成本较低，高优先级任务则要付出更高成本。只要还有空闲计算容量，就可以运行低优先级任务；但如果一项高优先级任务到来，低优先级任务随时可能遭到抢占。 这类价格较低的低优先级虚拟机，在 Amazon EC2 中称为 竞价实例（spot instance），在 Azure 中称为 竞价虚拟机（spot virtual machine），在 Google Cloud 中则称为 可抢占实例（preemptible instance）。
批处理通常用于时效性要求不高的作业，因此很适合使用低优先级任务和竞价实例，以降低运行成本。实质上，这些作业利用了原本会被闲置的计算资源，从而提高集群利用率。不过，这也意味着调度器更可能终止这些任务：抢占发生的频率要高于硬件故障。
由于批处理作业每次运行都从头生成输出，任务失败要比在线系统中的故障容易处理：系统可以删除失败执行留下的不完整输出，再把任务安排到另一台机器上重新运行。不过，只因一项任务失败就重跑整个作业会非常浪费。因此，MapReduce 及其后继系统让并行任务彼此独立，从而可以按单项任务的粒度重试工作。
如果一项任务的输出要在工作流中成为另一项任务的输入，容错就会更加棘手。MapReduce 的办法是始终把这类中间数据写回分布式文件系统，并等待写入任务顺利完成，之后才允许其他任务读取数据。即使在抢占频繁的环境中，这种做法也能正常工作，但它要向分布式文件系统写入大量数据，效率可能很低。
Spark 把中间数据保存在内存中，必要时“溢写”到本地磁盘，只把最终结果写入分布式文件系统。它还会跟踪中间数据的计算过程，以便在数据丢失时重新计算。Flink 采用另一种方法，定期为任务状态的快照建立检查点。
常见的两种模型是 MapReduce 和 数据流引擎。在实践中，数据流引擎已经基本取代 MapReduce，但理解 MapReduce 的工作原理仍然很有用，因为许多现代批处理框架都深受它的影响。
MapReduce 和数据流引擎已经逐渐支持多种编程模型，其中包括底层编程 API、关系查询语言和数据框 API。
MapReduce 的数据处理模式与“简单日志分析”中的 Web 服务器日志示例非常相似：
\n
是记录分隔符）。在 Hadoop MapReduce 中，输入文件存储在 HDFS
之类的分布式文件系统，或 S3
之类的对象存储中。系统可以使用多种文件格式，例如 Apache
Parquet（列式格式）或 Apache Avro（行式格式）。awk '{print $7}'：它提取
URL（$7）作为键，并把值留空。sort
命令完成。uniq -c
命令实现，它负责统计具有相同键的相邻记录数。要创建一项 MapReduce 作业，需要实现 mapper 和 reducer 两个回调函数，其行为如下：
Mapper:
每条输入记录都会调用一次 mapper，其任务是从输入记录中提取键和值。对于每条输入，它可以生成任意数量的键值对（也可以一个都不生成）。mapper 不会把某条输入记录的状态保留到下一条，因此每条记录都能独立处理。
Reducer:
MapReduce 框架接收 mapper 产生的键值对，收集属于同一个键的所有值，再用一个遍历该值集合的迭代器调用 reducer。reducer 可以生成输出记录，例如同一 URL 的出现次数。
实际上，使用原始 MapReduce API 实现复杂的处理作业非常困难而且费力——例如，作业使用的任何连接算法都必须从头实现。与较新的批处理器相比，MapReduce 的速度也很慢。其中一个原因是，基于文件的 I/O 无法形成作业流水线：上游作业完成之前，下游作业不能开始处理它的输出。
为了解决 MapReduce 的一些问题，人们开发了几种新的分布式批处理执行引擎，其中最著名的是 Spark 和 Flink 。它们的设计方式各有不同，却有一个共同点：把整个工作流作为一项作业处理，而不是拆成彼此独立的子作业。
这些系统显式建模了数据流经多个处理阶段的过程，因此称为 数据流引擎（dataflow engine）。与 MapReduce 一样，它们提供底层 API，通过反复调用用户定义的函数，每次处理一条记录；同时也提供 连接（join）和 分组（grouping）等高层算子。它们通过对输入分片来并行执行工作，并把一项任务的输出复制到另一项任务，作为后者的输入；如果两项任务运行在不同机器上，复制就通过网络完成。与 MapReduce 不同，算子不必严格交替扮演 map 和 reduce 的角色，而可以以更加灵活的方式组合。
这些数据流 API 通常使用关系模型风格的构件来表达计算：按照某个字段的值连接数据集，按键对元组分组，根据某项条件过滤数据，以及通过计数、求和等函数聚合元组。
这类处理引擎建立在 Dryad 和 Nephele 等研究系统的基础之上。与 MapReduce 模型相比，它们有几个优点：
数据流引擎能够实现与 MapReduce 工作流相同的计算，而且得益于上述优化，执行速度通常快得多。
开头的 Unix 工具示例和 MapReduce 都以排序为基础。批处理器必须能够对 PB 级的数据集排序，而这些数据不可能装入一台机器。因此，它们需要一种输入和输出都经过分片的分布式排序算法，这种算法称为 混洗（shuffle） 。
混洗是批处理器的一项基础算法，连接与聚合都要使用它。MapReduce、Spark、Flink、Daft、Dataflow 和 BigQuery 都实现了可伸缩的高性能混洗算法，以处理大型数据集。下面以 Hadoop MapReduce 中的混洗为例，不过本节的概念也适用于其他系统。
下面展示了一项 MapReduce 作业中的数据流，假设作业的输入已经分片，三个分片分别标为 m1、m2、m3。例如，每个分片可以是 HDFS 中的单独文件，也可以是对象存储中的单独对象。同一数据集的全部分片，可以集中放在同一个 HDFS 目录中；在对象存储的存储桶里，它们也可以使用相同的键前缀。
框架会为每个输入分片启动一项单独的 map 任务。任务读取分配给它的文件，每次把一条记录传给 mapper 回调函数。计算的 reduce 端同样经过分片。map 任务的数量由输入分片数决定，而 reduce 任务的数量则由作业作者配置，两者可以不同。
mapper 的输出由键值对组成。框架必须保证：如果两个不同的 mapper 输出了相同的键，这些键值对最终会由同一个 reducer 任务处理。 为此，每个 mapper 都会在本地磁盘上为每个 reducer 分别创建一个输出文件。
mapper 在写入这些文件的同时，还会在每个文件中按键排列键值对。先在内存的有序数据结构中收集一批键值对，再把它们写成有序段文件，然后逐步将较小的段合并成较大的段。
每个 mapper 完成后，reducer 会连接到它，并把属于自己的有序键值对文件复制到本地磁盘。reduce 任务取得所有 mapper 输出中属于自己的那一份后，会像归并排序一样合并这些文件，同时保持排序顺序。这样一来，即使具有相同键的键值对来自不同的 mapper，最终也会彼此相邻。随后，每个键都会调用一次 reducer 函数，并向它传入一个迭代器，用于返回该键对应的所有值。
reducer 函数产生的记录会顺序写入文件，每项 reduce 任务对应一个文件。图 11-1中的 r 1、r 2 和 r 3 就构成了作业输出数据集的三个分片，它们会被写回分布式文件系统或对象存储。
MapReduce 在 map 与 reduce 阶段之间执行混洗，而现代数据流引擎和云数据仓库则要精巧得多。BigQuery 等系统优化了混洗算法，尽量把数据保存在内存中，或把数据写入外部排序服务。这类服务既能加快混洗，又能通过复制混洗数据来提高容错能力。
下面来看有序数据如何简化分布式连接与聚合。为便于说明，我们仍以 MapReduce 为例，不过这些概念适用于大多数批处理系统。
展示了批处理作业中一个典型的连接示例。左侧是一份事件日志，记录已登录用户在网站上的操作，这些记录称为 活动事件（activity event），也叫 点击流数据（clickstream data）；右侧则是用户数据库。这个例子可以看作星型模式的一部分：事件日志是事实表，用户数据库则是其中一张维度表。
如果想结合用户数据库中的信息来分析活动事件，例如利用用户资料中的出生日期，了解某些页面更受年轻用户还是年长用户欢迎，就需要连接这两张表。假设两张表都大到必须分片，该如何计算这种连接？
可以利用 MapReduce 的一个性质：无论键值对最初位于哪个分片，混洗都会把具有相同键的键值对集中到同一个 reducer。这里可以把用户 ID 作为键。因此，我们可以编写一个 mapper 遍历用户活动事件，以用户 ID 为键，输出页面浏览的 URL，如图 11-3 所示。另一个 mapper 逐行遍历用户数据库，提取用户 ID 作为键、用户出生日期作为值。
混洗随后确保 reducer 函数能够同时访问某位用户的出生日期，以及该用户的所有页面浏览事件。MapReduce 甚至可以安排记录的顺序，让 reducer 总是先看到用户数据库中的记录，接着再按时间戳顺序看到活动事件；这种技术称为 二次排序（secondary sort）。
这样，reducer 就能轻松完成实际的连接逻辑。第一个值应当是出生日期，reducer 先把它保存在局部变量中，再遍历具有相同用户 ID 的活动事件，输出每个浏览过的 URL 以及浏览者的出生日期。reducer 会一次处理某个用户 ID 的全部记录，所以任何时刻只需在内存中保存一条用户记录，也完全不必发出网络请求。这种算法称为 排序合并连接（sort-merge join），因为 mapper 的输出按键排序，而 reducer 随后会合并连接两侧的有序记录列表。
工作流中的下一项 MapReduce 作业可以继续计算每个 URL 的浏览者年龄分布。它首先用 URL 作为键混洗数据；排序完成后，reducer 会遍历同一个 URL 的所有页面浏览记录（其中包含浏览者的出生日期），为各个年龄段维护浏览次数计数器，并为每条页面浏览记录递增相应的计数器。这样就实现了 分组（group by）操作和聚合。
多年来，分布式批处理的执行引擎日趋成熟。如今，基础设施已经足够稳健，可以在超过 10,000 台机器的集群上存储和处理数 PB 数据。既然如此规模的批处理系统如何实际运行已经大体得到解决，人们就把注意力转向了编程模型的改进。
MapReduce、数据流引擎和云数据仓库都采用 SQL 作为批处理的通用语言。这是很自然的选择：传统数据仓库本来就使用 SQL，数据分析和 ETL 工具也已经支持 SQL，而且开发者和分析师无不熟悉 SQL。
与手写 MapReduce 作业相比，查询语言接口除了所需代码更少，还有一个明显的优点：它们可以交互使用。你可以编写分析查询，再从终端或图形界面运行。这种交互查询方式很高效也很自然，业务分析师、产品经理、销售和财务团队等人员都可以借此在批处理环境中探索数据。尽管它不是经典形式的批处理，SQL 支持仍然使探索性查询成为分布式批处理系统的一项适用场景。
高级查询语言不仅提高了使用系统的人所能达到的生产率，也能从机器层面提高作业执行效率。查询引擎负责把 SQL 查询转换成在集群中执行的批处理作业。从查询转换到语法树、再转换成物理算子的过程，让引擎有机会优化查询。Hive、Trino、Spark 和 Flink 等查询引擎都提供基于代价的查询优化器，可以分析连接输入的特征，自动决定哪一种算法最适合眼前的任务。优化器甚至可以改变连接的顺序，以尽量减少中间状态。
SQL 是最流行的通用批处理查询语言，不过其他语言仍用于一些专门场景。Apache Pig 是一种以关系算子为基础的语言，允许用户逐步描述数据管道，而不是把所有逻辑写成一个庞大的 SQL 查询。数据框 具有类似特征，Morel 则是受 Pig 影响的一种较新语言。还有一些用户采用 jq、JMESPath 或 JsonPath 等 JSON 查询语言。许多图处理框架也支持通过查询语言进行批计算，例如 Apache TinkerPop 的 Gremlin。
随着数据科学家和统计学家开始使用 分布式批处理框架 进行 机器学习，他们发现现有的处理模型用起来很麻烦，因为他们习惯的是 R 和 pandas 中的 数据框模型（见 数据模型与查询语言 章节）。
数据框与关系数据库中的表相似：它由许多行组成，同一列中的所有值都具有相同类型。用户不必编写一条庞大的 SQL 查询，而是调用与关系算子对应的函数，执行过滤、连接、排序、分组等操作。
最初，数据框操作一般在本机内存中执行，因而只能处理单台机器能够容纳的数据集。数据科学家希望继续使用熟悉的数据框 API，与批处理环境中的大型数据集交互。Spark、Flink 和 Daft 等分布式数据处理框架为满足这种需求，也采用了数据框 API。不过，本地数据框通常带有索引并且有序，分布式数据框一般却不具备这些性质。因此，把程序迁移到批处理框架后，其性能可能出人意料。
数据框 API 看起来与数据流 API 相似，但具体实现各不相同。pandas 会在数据框方法被调用时立即执行操作；Apache Spark 则会先把所有数据框 API 调用转换成查询计划，经过查询优化，再在分布式数据流引擎上执行工作流。这样便能改善性能。
Daft 等框架甚至同时支持客户端与服务端计算：规模较小的内存操作在客户端执行，规模较大的数据集和计算则在服务端执行。Apache Arrow 等列式存储格式提供了统一的数据模型，可由客户端与服务端的执行引擎共享。
批处理在各种应用中怎样发挥作用。批处理作业非常适合成批处理大型数据集，却不适合低延迟场景。因此，只要数据量很大、数据新鲜度又不重要，通常就能看到批处理作业。听起来这似乎是一个很大的局限，但事实证明，相当多的数据处理都符合这一模型：
ETL 和 ELT：数据处理管道从生产数据库抽取数据，对其进行转换，再把结果加载到下游系统。本节用“ETL”同时指代 ETL 与 ELT 工作负载。这类工作负载经常由批处理作业完成，尤其是在下游系统为数据仓库时。
批处理作业天然具有并行性，因而非常适合数据转换。许多数据转换工作负载都易于并行：过滤数据、投影字段，以及其他许多常见的数据仓库转换，都可以并行完成。
批处理环境还配有稳健的工作流调度器，能够轻松安排、编排和调试 ETL 数据管道作业。发生故障时，调度器通常会重试作业，以消除可能出现的暂时性问题。作业若反复失败，就会被标记为失败，开发者可以很容易地看到数据管道中的哪项作业停止了工作。Airflow 等调度器甚至内置了 MySQL、PostgreSQL、Snowflake、Spark 和 Flink 等数十种流行系统的数据源、数据接收端及查询算子。调度器与数据处理系统的紧密集成简化了数据集成。
批处理作业出了问题之后很容易排查和修复；这一点在调试数据管道时尤其有用。可以直接检查有问题的文件，找出错误所在；修复 ETL 批处理作业后，再重新运行即可。例如，输入文件可能不再包含转换作业所要使用的某个字段。数据工程师发现字段缺失后，可以更新转换逻辑，或者修改产生该输入的作业。
过去，数据管道通常由一支数据工程团队统一管理，因为要求开发产品功能的其他团队编写和管理复杂的批处理数据管道并不公平。近来，批处理模型和元数据管理的改进，使组织中的工程师更容易参与和管理自己的数据管道。数据网格（data mesh）、数据契约（data contract）和 数据编织（data fabric）等实践提供了标准与工具，帮助团队安全地发布数据，供组织中的任何人使用。
如今，数据管道与分析查询不仅开始共享处理模型，也开始共享执行引擎。许多批处理 ETL 作业与读取其输出的分析查询，会运行在同一套系统上。数据管道转换和分析查询都以 SparkSQL、Trino 或 DuckDB 查询来执行，已经十分常见。这样的架构进一步模糊了应用工程、数据工程、分析工程和业务分析之间的界限。
在“分析型与事务型系统”中，我们看到分析查询（OLAP）经常扫描大量记录，并执行分组与聚合。这样的工作负载可以与其他批处理工作负载一起，在批处理系统中运行。分析师编写 SQL 查询，由查询引擎执行，并读写分布式文件系统或对象存储。表与文件之间的映射、名称和类型等表元数据，则通过 Apache Iceberg 等表格式和 Unity 等目录服务管理。这种架构称为 数据湖仓（data lakehouse）。
与 ETL 一样，SQL 查询接口的改进使许多组织如今也使用 Spark 等批处理框架进行分析。这类查询模式分为两种：
SQL 支持还让批处理框架能够与电子表格和 Tableau、Power BI、Looker、Apache Superset 等数据可视化工具集成。例如，Tableau 提供 SparkSQL 和 Presto 连接器；Apache Superset 则支持 Trino、Hive、Spark SQL、Presto 等许多最终会执行批处理作业来查询数据的系统。
机器学习（ML）经常用到批处理。数据科学家、机器学习工程师和 AI 工程师使用批处理框架来探索数据规律、转换数据并训练机器学习模型。常见用途包括：
批处理框架为这些场景提供了专门的工具。例如，Apache Spark 的 MLlib 和 Apache Flink 的 FlinkML 都内置了丰富的特征工程工具、统计函数和分类器。
推荐引擎、排序系统等机器学习应用也大量使用图处理。许多图算法可以表述为：每次沿一条边遍历，将一个顶点与相邻顶点连接起来以传播某些信息，并不断重复，直到满足某项条件——例如没有更多的边可以继续遍历，或者某项指标已经收敛。
批量同步并行（bulk synchronous parallel，BSP）计算模型 已经成为批量处理图数据时的常用模型。Apache Giraph、Spark 的 GraphX API 和 Flink 的 Gelly API 等系统都实现了这一模型。它也称为 Pregel 模型，因为 Google 的 Pregel 论文推广了这种图处理方法。
批处理也是 大语言模型（LLM）数据准备与训练的重要组成部分。网站等原始文本输入通常存放在分布式文件系统或对象存储中，必须先经过预处理才能用于训练。适合交给批处理框架完成的预处理步骤包括：
Kubeflow、Flyte 和 Ray 等批处理框架就是为这类工作负载设计的。例如，OpenAI 在 ChatGPT 的训练过程中就使用了 Ray。这些框架内置了与 PyTorch、TensorFlow、XGBoost 等大语言模型和 AI 库的集成，并直接支持特征工程、模型训练、批量推理和微调（针对特定用例调整基础模型）等操作。
最后，数据科学家还经常在 Jupyter 或 Hex 等交互式笔记本中对数据进行实验。笔记本由若干 单元格（cell）组成，每个单元格都是一小段 Markdown、Python 或 SQL。按顺序执行这些单元格，可以生成电子表格、图表或数据。许多笔记本通过数据框 API 使用批处理系统，或者用 SQL 查询这类系统。
批处理作业经常用于构建预计算或衍生数据集，例如商品推荐、面向用户的报表，以及机器学习模型所需的特征。这些数据集通常由生产数据库、键值存储或搜索引擎对外提供。不论使用哪种系统，预计算数据最终都必须从批处理系统的分布式文件系统或对象存储，回到承载线上流量的数据库中。
最直观的选择或许是直接在批处理作业中调用数据库客户端库，逐条写入数据库服务器。这种方法确实能用——前提是防火墙规则允许批处理环境直接访问生产数据库——但基于以下几个原因，它并不是个好主意：
更好的方案是让批处理作业把预计算数据集推送到 Kafka 主题之类的流中，Elasticsearch 等搜索引擎、Apache Pinot 和 Apache Druid 等实时 OLAP 系统、Venice 等衍生数据存储，以及 ClickHouse 等云数据仓库，都内置了从 Kafka 摄取数据的能力。让数据经过流系统，可以缓解上述问题中的一部分：
不过，让数据经过流并不会自动解决前面提到的“全有或全无”保证问题。为此，批处理作业必须在完成时通知下游系统：作业已经完成，数据现在可以对外提供。流消费者则必须能够在收到完成通知之前，让接收到的数据对查询保持不可见，就像采用 读已提交（read committed）隔离级别时未提交的事务一样。
另一种模式在数据库初始化时更为常见：直接在批处理作业 内部 构建一个全新的数据库，再把分布式文件系统、对象存储或本地文件系统中的文件批量加载到该数据库中。许多数据系统都提供了相应的批量导入工具，例如 TiDB 的 Lightning 工具，以及 Apache Pinot 和 Apache Druid 的 Hadoop 导入作业。RocksDB 也提供了从批处理作业批量导入 SST 的 API。
通过批处理构建数据库并批量导入数据，速度非常快，也更便于系统在不同数据集版本之间进行原子切换。另一方面，由批处理作业构建全新数据库时，要增量更新数据集会比较困难。如果同时需要初始化和增量加载，通常会采用混合方案。例如，Venice 支持混合存储，既可以进行基于行的批量更新，也可以切换整个数据集。