---
title: DDIA 2-е изд. (Kleppmann & Riccomini, «Designing Data-Intensive Applications», 2nd ed.) — глава 11: Batch Processing
source: materials/DDIA-2nd-edition.pdf, стр. 475–510
конспект: TASK-45.39, извлечено pymupdf 2026-09-17; ниже полный «грязный» текст главы + выжимка
приоритет: опционально
статус: выжимка перенесена в knowledge-base.md / ddia-map.md — см. _map.md
---

# Глава 11. Пакетная обработка (Batch Processing)

## Выжимка

1. Online (запрос-ответ, латентность) vs offline/**batch** (bounded вход → derived output с нуля, детерминированность позволяет перезапускать); Unix-философия «маленькие программы + pipes» как прообраз MapReduce-конвейеров.
2. MapReduce → dataflow-движки (Spark и др.): фиксация графа вычислений, оптимизация, избегание materialization между этапами.
3. Join'ы в batch: reduce-side (партиционирование по ключу обеих сторон), map-side/broadcast (маленькая таблица в память), hash join — выбор по размерам и паттерну данных.
4. Batch строит производные данные: поисковые индексы, рекомендации, обучение ML, аналитика; reprocess = rebuild представления из лога.
5. Выход — в object storage/олап; слияние batch и stream в unified-движках — мост к гл. 12–13.

## Полный текст (грязная выгрузка)

CHAPTER 11
Batch Processing
A system cannot be successful if it is too strongly influenced by a single person. Once the
initial design is complete and fairly robust, the real test begins as people with many different
viewpoints undertake their own experiments.
—Donald Knuth, “The Errors of TeX” (1989)
Much of this book so far has talked about requests and queries and the corresponding
responses or results. This style of data processing is assumed in many modern data
systems: you ask for something, or you send an instruction, and the system tries to
give you an answer as quickly as possible.
A web browser requesting a page, a service calling a remote API, databases, caches,
search indexes, and many other systems work this way. We call these online systems.
Response time is usually their primary measure of performance, and they often
require fault tolerance to ensure high availability.
However, sometimes you need to run a bigger computation or process larger amounts
of data than you can do in an interactive request. Maybe you need to train an AI
model, or transform lots of data from one form into another, or compute analytics
over a very large dataset. We call these tasks batch processing jobs, and the systems
that handle them are sometimes referred to as offline systems.
A batch processing job takes input data (which is read-only) and produces output
data (which is generated from scratch every time the job runs). It typically does
not mutate data in the way a read/write transaction would. The output is therefore
derived from the input (as discussed in “Systems of Record and Derived Data” on
page 10). If you don’t like the output, you can delete it, adjust the job’s logic, and run
the job again.
451


By treating inputs as immutable and avoiding side effects (such as writing to external
databases), batch jobs achieve good performance as well as other benefits:
• If you introduce a bug into the code and the output is wrong or corrupted, you
•
can simply roll back to a previous version of the code and rerun the job, and
the output will be correct again. Or, even simpler, you can keep the old output
in a different directory and switch back to it. Most object stores and open table
formats (see “Cloud Data Warehouses” on page 135) support this feature, which
is known as time travel. Most databases with read/write transactions do not have
this property: if you deploy buggy code that writes bad data to the database,
rolling back the code will do nothing to fix that data. The idea of being able to
recover from buggy code has been called human fault tolerance [1].
• As a consequence of this ease of rolling back, feature development can proceed
•
more quickly than in an environment where mistakes could mean irreversible
damage. This principle of minimizing irreversibility is beneficial for Agile soft‐
ware development [2].
• The same set of files can be used as input for various types of jobs, including
•
monitoring jobs that calculate metrics and evaluate whether a job’s output has
the expected characteristics (for example, by comparing it to the output from the
previous run and measuring discrepancies).
• Batch processing frameworks make efficient use of computing resources. Even
•
though it’s possible to batch-process data via online data systems such as OLTP
databases and application servers, doing so can be much more expensive in terms
of the resources required.
Batch processing has proven useful in a wide range of use cases, which we’ll revisit
in “Batch Use Cases” on page 476. However, it also presents challenges. With most
frameworks, output can be processed by other jobs only after the whole job finishes.
Batch processing can also be inefficient; any change to the input data—even a single
byte—requires the batch job to reprocess the entire input dataset.
A batch job may take a long time to run: minutes, hours, or even days. Jobs may
be scheduled to run periodically (e.g., once per day). The primary measure of perfor‐
mance is usually throughput: how much data the job can process per unit of time.
Some batch systems handle faults by simply aborting and restarting the whole job,
while others have fault tolerance so that a job can complete successfully despite some
of its nodes crashing.
The boundary between online and batch processing systems is not always clear;
a long-running database query looks quite a bit like a batch process. But batch
processing also has particular characteristics that make it a useful building block for
building reliable, scalable, and maintainable applications. For example, it often plays a
role in data integration—composing multiple data systems to achieve things that one
452 
| 
Chapter 11: Batch Processing


system alone cannot do. ETL, as discussed in “Data Warehousing” on page 7, is an
example of this.
An alternative to batch processing is stream processing, in which
the job doesn’t finish running when it has processed the input, but
instead continues watching the input and processes changes in the
input shortly after they happen. We will turn to stream processing
in Chapter 12.
Modern batch processing has been heavily influenced by MapReduce, a batch pro‐
cessing algorithm that was published by Google in 2004 [3] and subsequently imple‐
mented in various open source data systems, including Hadoop, CouchDB, and
MongoDB. MapReduce is a fairly low-level programming model, less sophisticated
than the parallel query execution engines found, for example, in data warehouses
[4, 5]. When it was new, MapReduce was a step forward in terms of the scale of
processing that could be achieved on commodity hardware, but now it is largely
obsolete and no longer used at Google [6, 7].
Batch processing today is more often done using frameworks such as Spark or Flink,
or data warehouse query engines. Like MapReduce, they rely heavily on sharding (see
Chapter 7) and parallel execution, but they have far more sophisticated caching and
execution strategies. As these systems have matured, operational concerns have been
largely solved, so focus has shifted toward usability. New processing models such as
dataflow APIs, query languages, and DataFrame APIs are now widely supported. Job
and workflow orchestration has also matured. Hadoop-centric workflow schedulers
such as Oozie and Azkaban have been replaced with more generalized solutions such
as Airflow, Dagster, and Prefect, which support a wide array of batch processing
frameworks and cloud data warehouses.
Cloud computing has grown ubiquitous. Batch storage layers are shifting from dis‐
tributed filesystems (DFSs) like HDFS (Hadoop Distributed File System), GlusterFS,
and CephFS to object storage systems such as S3. Scalable cloud data warehouses like
BigQuery and Snowflake are blurring the line between data warehouses and batch
processing.
To build an intuition of what batch processing is about, we will start this chapter
with an example that uses standard Unix tools on a single machine. We will then
investigate how we can extend data processing to multiple machines in a distributed
system. We will see that, much like an operating system, distributed batch processing
frameworks have a scheduler and a filesystem. We will then explore various process‐
ing models that we use to write batch jobs. Finally, we will discuss common batch
processing use cases.
Batch Processing 
| 
453


Batch Processing with Unix Tools
Say you have a web server that appends a line to a log file every time it serves a
request. For example, using the NGINX default access log format, one line of the log
might look like this:
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"
(That is actually one line; it’s split into multiple lines here for readability.) There’s a lot
of information in that line. To interpret it, you need to look at the definition of the log
format, which is as follows:
$remote_addr - $remote_user [$time_local] "$request"
$status $body_bytes_sent "$http_referer" "$http_user_agent"
So, this one line of the log indicates that on June 27, 2025, at 17:55:11 UTC, the
server received a request for the file /css/typography.css from the client IP address
216.58.210.78. The user was not authenticated, so $remote_user is set to a hyphen
(-). The response status was 200 (i.e., the request was successful), and the response
was 3,377 bytes in size. The web browser was Chrome 137, and it loaded the file
because it was referenced in the page at the URL https://martin.kleppmann.com/.
Though log parsing might seem contrived, it’s a critical part of the operations of
many modern technology companies and is used for everything from ad pipelines
to payment processing. Indeed, it was a driving force behind the rapid adoption of
MapReduce and the “big data” movement.
Simple Log Analysis
Various tools can take these log files and produce pretty reports about your website
traffic, but for the sake of exercise, let’s build our own, using basic Unix tools. For
example, say you want to find the five most popular pages on your website. You can
do this in a Unix shell as follows: 
cat /var/log/nginx/access.log | 
  awk '{print $7}' | 
  sort             | 
  uniq -c          | 
  sort -r -n       | 
  head -n 5          
Read the log file. (Strictly speaking, cat is unnecessary here, as the input file
could be given directly as an argument to awk. However, the linear pipeline is
more apparent when it’s written like this.)
454 
| 
Chapter 11: Batch Processing


Split each line into fields by whitespace, and output only the seventh field from
each line, which happens to be the requested URL. In our example line, this URL
is /css/typography.css.
Alphabetically sort the list of requested URLs. The reason for sorting is to
ensure that, if a URL has been requested n times, the sorted file will contain the
same URL repeated n times in a row.
The uniq command filters out repeated lines in its input by checking whether
two adjacent lines are the same. The -c option tells it to also output a counter: for
every distinct URL, it reports how many times that URL appeared in the input.
The second sort sorts by the number (-n) at the start of each line, which is the
number of times the URL was requested. It then returns the results in reverse
(-r) order—with the largest number first.
Finally, head outputs just the first five lines (-n 5) of input and discards the rest.
The output of that series of commands looks something like this:
4189 /favicon.ico
3631 /2016/02/08/how-to-do-distributed-locking.html
2124 /2020/11/18/distributed-systems-and-elliptic-curves.html
1369 /
 915 /css/typography.css
Although the preceding command line likely looks a bit obscure if you’re unfamiliar
with Unix tools, it is incredibly powerful. It will process gigabytes of log files in a
matter of seconds, and you can easily modify the analysis to suit your needs. For
example, if you want to omit CSS files from the report, you can change the awk
argument to $7 !~ /\.css$/ {print $7}, and if you want to count top client IP
addresses instead of top pages, you can change the awk argument to {print $1}, and
so on.
We don’t have space in this book to explore Unix tools in detail, but they are very
much worth learning about. Many data analyses can be done in a few minutes using a
combination of awk, sed, grep, sort, uniq, and xargs, and they perform surprisingly
well [8].
Batch Processing with Unix Tools 
| 
455


Chain of Commands Versus Custom Program
Instead of the chain of Unix commands, you could write a simple program to do the
same thing. For example, in Python, it might look something like this:
from collections import defaultdict
counts = defaultdict(int) 
with open('/var/log/nginx/access.log', 'r') as file:
    for line in file:
        url = line.split()[6] 
        counts[url] += 1 
top5 = sorted(((count, url) for url, count in counts.items()),
              reverse=True)[:5] 
for count, url in top5:  
    print(f"{count} {url}")
Initialize counts to be a hash table that keeps a counter for the number of times
we’ve seen each URL. The initial value of each counter is 0.
Get the requested URL, which is the seventh whitespace-separated field, from
each line of the log (the array index is 6 because Python’s arrays are zero-
indexed).
Increment the counter for the URL in the current line of the log.
Sort the hash table contents by counter value (descending), and take the top five
entries.
Print out those top five entries.
This program is not as concise as the chain of Unix commands, but it’s fairly read‐
able, and which of the two you prefer is in part a matter of taste. However, besides
the superficial syntactic differences between the two, there is a big difference in the
execution flow, which becomes apparent if you run this analysis on a large file.
Sorting Versus In-Memory Aggregation
The Python script keeps an in-memory hash table of URLs, where each URL is
mapped to the number of times it has been seen. The Unix pipeline example does not
have such a hash table, but instead relies on sorting a list of URLs in which multiple
occurrences of the same URL are simply repeated.
456 
| 
Chapter 11: Batch Processing


Which approach is better? It depends on the number of URLs you have. For most
small to mid-sized websites, you can probably fit all distinct URLs, and a counter
for each URL, in (say) 1 GB of memory. In this example, the working set of the job
(the amount of memory to which the job needs random access) depends only on the
number of distinct URLs. If there are a million log entries for a single URL, the space
required in the hash table is still just one URL plus the size of the counter. If this
working set is small enough, an in-memory hash table works fine—even on a laptop.
On the other hand, if the job’s working set is larger than the available memory, the
sorting approach has the advantage that it can make efficient use of disks. It’s the
same principle we discussed in “Log-Structured Storage” on page 118: chunks of data
can be sorted in memory and written out to disk as segment files, and then multiple
sorted segments can be merged into a larger sorted file. Mergesort has sequential
access patterns that perform well on disks (see “Sequential Versus Random Writes on
SSDs” on page 130).
The sort utility in GNU Coreutils (Linux) automatically handles larger-than-
memory datasets by spilling to disk and automatically parallelizes sorting across
multiple CPU cores [9]. This means that the simple chain of Unix commands we saw
earlier easily scales to large datasets, without running out of memory. The bottleneck
is likely to be the rate at which the input file can be read from disk.
A limitation of Unix tools is that they run on a single machine. Datasets that are
too large to fit in memory or on the local disk present a problem—and that’s where
distributed batch processing frameworks come in.
Batch Processing in Distributed Systems
The machine that runs our Unix tools example has a number of components that
work together to process the log data:
• Storage devices that are accessed through the operating system’s filesystem
•
interface
• A scheduler that determines when processes get to run and how to allocate CPU
•
resources to them
• A series of Unix programs whose standard input and standard output (stdin and
•
stdout) are connected together by pipes
These same components exist in distributed data processing frameworks. In fact, you
can think of these frameworks as distributed operating systems; they have filesystems,
job schedulers, and programs that send data to one another through the filesystem or
other communication channels.
Batch Processing in Distributed Systems 
| 
457


Distributed Filesystems
The filesystem provided by your operating system is composed of several layers:
• At the lowest level, block device drivers speak directly to the disk and allow the
•
layers above to read and write raw blocks.
• Above the block layer sits a page cache that keeps recently accessed blocks in
•
memory for faster access.
• The block API is wrapped in a filesystem layer that breaks large files into blocks
•
and tracks file metadata such as inodes, directories, and files. Two common
implementations on Linux are ext4 and XFS, for example.
• Finally, the operating system exposes different filesystems to applications through
•
a common API called the virtual filesystem (VFS). The VFS is what allows
applications to read and write in a standard way regardless of the underlying
filesystem.
Distributed filesystems work in much the same way. Files are broken into blocks,
which are distributed across many machines. DFS blocks are typically much larger
than local blocks. HDFS defaults to 128 MB, while JuiceFS and many object stores use
4 MB blocks—much larger than ext4’s 4,096 bytes. Larger blocks mean less metadata
to keep track of, which makes a big difference on petabyte-sized datasets. Larger
blocks also lower the overhead of seeking to a block relative to reading it.
Most physical storage devices can’t write partial blocks, so operating systems require
writes to use an entire block even if the data doesn’t take up the whole block. Since
distributed filesystems have larger blocks and are usually implemented on top of
operating system filesystems, they don’t have this requirement. For example, a 900
MB file stored with 128 MB blocks would have seven blocks that use 128 MB and one
block that uses 4 MB.
DFS blocks are read by making network requests to a machine in the cluster that
stores the block. Each machine runs a daemon, exposing an API that allows remote
processes to read and write blocks as files on its local filesystem. HDFS refers to these
daemons as DataNodes, while GlusterFS calls them glusterfsd processes. We’ll call
them data nodes in this book.
Distributed filesystems also implement the distributed equivalent of a page cache.
Since DFS blocks are stored as files on data nodes, reads and writes go through
each data node’s operating system, which includes an in-memory page cache. This
keeps frequently read data blocks in memory on the data nodes. Some distributed
filesystems also implement more caching tiers, such as the client-side and local disk
caching found in JuiceFS.
458 
| 
Chapter 11: Batch Processing


Filesystems such as ext4 and XFS keep track of storage metadata including free space,
file block locations, directory structures, permission settings, and more. Distributed
filesystems also need a way to track file locations spread across machines, permission
settings, and so on. Hadoop has a service called the NameNode that maintains
metadata for the cluster. DeepSeek’s 3FS has a metadata service that persists its data to
a key-value store such as FoundationDB.
Above the filesystem sits the VFS. A close analogue in batch processing is a dis‐
tributed filesystem’s protocol. Distributed filesystems must expose a protocol or inter‐
face so that batch processing systems can read and write files. This protocol acts as
a pluggable interface; any DFS may be used so long as it implements the protocol.
For example, Amazon S3’s API has been widely adopted by storage systems such as
MinIO, Cloudflare’s R2, Tigris, Backblaze’s B2, and many others. Batch processing
systems with S3 support can use any of these storage systems.
Some DFSs implement POSIX-compliant filesystems that appear to the operating sys‐
tem’s VFS like any other filesystem. Filesystem in Userspace (FUSE) or the Network
File System (NFS) protocol are often used to integrate into the VFS. NFS is perhaps
the most well-known distributed filesystem protocol. The protocol was originally
developed to allow multiple clients to read and write data on a single server. More
recently, filesystems such as Amazon Elastic File System (EFS) and Archil provide
NFS-compatible distributed filesystem implementations that are far more scalable.
NFS clients still connect to one endpoint, but underneath, these systems communi‐
cate with distributed metadata services and data nodes to read and write data.
Distributed Filesystems and Network Storage
Distributed filesystems are based on the shared-nothing principle (see “Shared-
Memory, Shared-Disk, and Shared-Nothing Architectures” on page 51), in contrast
to the shared-disk approach of network attached storage (NAS) and storage area
network (SAN) architectures. Shared-disk storage is implemented by a centralized
storage appliance, often using custom hardware and special network infrastructure
such as Fibre Channel. On the other hand, the shared-nothing approach requires no
special hardware, only computers connected by a conventional datacenter network.
Many distributed filesystems are built on commodity hardware, which is less expen‐
sive but has higher failure rates than enterprise-grade hardware. In order to tolerate
machine and disk failures, file blocks are replicated on multiple machines. This also
allows schedulers to more evenly distribute workloads since they can execute a task
on any node that contains a replica of the task’s input data.
Replication may mean simply keeping several copies of the same data on multiple
machines, as described in Chapter 6, or using an erasure coding scheme such as
Batch Processing in Distributed Systems 
| 
459


Reed–Solomon codes, which allows lost data to be recovered with lower storage
overhead than full replication [10, 11, 12]. The techniques are similar to RAID,
which provides redundancy across several disks attached to the same machine; the
difference is that in a distributed filesystem, file access and replication are done over a
conventional datacenter network without special hardware. 
Object Stores
Object storage services such as Amazon S3, Google Cloud Storage, Azure Blob
Storage, and OpenStack Swift have become a popular alternative to distributed file‐
systems for batch processing jobs. In fact, the line between the two is somewhat
blurry. As we saw in the previous section and “Databases Backed by Object Storage”
on page 202, FUSE drivers allow users to treat object stores such as S3 as a filesystem.
Some DFS implementations, such as JuiceFS and Ceph, offer both object storage and
filesystem APIs. However, their APIs, performance, and consistency guarantees are
very different. Care must be taken when adopting such systems to make sure they
behave as expected, even if they seem to implement the requisite APIs.
Each object in an object store has a URL such as s3://my-photo-bucket/2025/04/01
/birthday.png. The host portion of the URL (my-photo-bucket) describes the bucket
where objects are stored, and the part that follows is the object’s key (/2025/04/01
/birthday.png in our example). A bucket has a globally unique name, and each object’s
key must be unique within its bucket.
Objects are read using a get call and written using a put call. Unlike files on a
filesystem, objects are immutable once written. To update an object, it must be fully
rewritten using a put call, similarly to a key-value store. Azure Blob Storage and S3
Express One Zone support appends, but most other stores do not. There are no file
handle APIs with functions like fopen and fseek.
Objects may look as if they are organized into directories, which is somewhat confus‐
ing because object stores do not have the concept of directories. The path structure
is simply a convention, and the slashes are a part of the object’s key. This convention
allows you to perform something similar to a directory listing by requesting a list of
objects with a particular prefix. However, listing objects by prefix is different from a
filesystem directory listing in two ways:
• A prefix list operation behaves like a recursive ls -R call on a Unix system. It
•
returns all objects that start with the prefix, and objects in subpaths are included.
• Empty directories are not possible. If you were to remove all objects underneath
•
s3://my-photo-bucket/2025/04/01, then 01 would no longer appear when you
called list on s3://my-photo-bucket/2025/04. It’s common practice to create a
zero-byte object as a way to represent an empty directory (e.g., creating an empty
460 
| 
Chapter 11: Batch Processing


s3://my-photo-bucket/2025/04/01 file to keep it present when all child objects are
deleted).
DFS implementations often support common filesystem operations such as hard
links, symbolic links, file locking, and atomic renames. Such features are missing
from object stores. Linking and locks are typically not supported, while renames
are nonatomic; they’re accomplished by copying the object to the new key and then
deleting the old object. If you want to rename a directory, you have to individually
rename every object within it, since the directory name is a part of the key.
The key-value stores we discussed in Chapter 4 are optimized for small values
(typically kilobytes) and frequent, low-latency reads/writes. In contrast, distributed
filesystems and object stores are generally optimized for large objects (megabytes to
gigabytes) and less frequent, larger reads. Recently, however, object stores have begun
to add support for frequent and smaller reads/writes. For example, S3 Express One
Zone now offers single-millisecond latency and a pricing model that is more similar
to key-value stores.
Another difference between distributed filesystems and object stores is that DFSs
such as HDFS allow computing tasks to be run on the machine that stores a copy
of a particular file. This allows the task to read that file without having to send
it over the network, which saves bandwidth if the executable code of the task is
smaller than the file it needs to read. On the other hand, object stores usually keep
storage and computation separate. Doing so might use more bandwidth, but modern
datacenter networks are very fast, so this is often acceptable. This architecture also
allows machine resources such as CPU and memory to be scaled independently of
storage since the two are decoupled. 
Distributed Job Orchestration
Our operating system analogy also applies to job orchestration. When you execute
a Unix batch job, something needs to actually run the awk, sort, uniq, and head
processes. Data needs to be transferred from one process’s output to another process’s
input, memory must be allocated for each process, instructions from each process
must be scheduled fairly and executed on the CPU, memory and I/O boundaries
must be enforced, and so on. On a single machine, an operating system’s kernel is
responsible for such work. In a distributed environment, this is the role of a job
orchestrator.
Batch processing frameworks send a request to an orchestrator’s scheduler to run a
job. Requests to start a job contain metadata such as this:
• The number of tasks to execute
•
• The amount of memory, CPU, and disk needed for each task
•
Batch Processing in Distributed Systems 
| 
461


• A job identifier
•
• Access credentials
•
• Job parameters such as input and output data
•
• Required hardware details such as GPUs or disk types
•
• The location of the job’s executable code
•
Orchestrators such as Kubernetes and Hadoop YARN (Yet Another Resource Nego‐
tiator) [13] combine this information with cluster metadata to execute the job using
the following components:
Task executors
An executor daemon such as YARN’s NodeManager or Kubernetes’s kubelet runs
on each node in the cluster. Executors are responsible for running job tasks,
sending heartbeats to signal their liveness, and tracking task status and resource
allocation on the node. When a task-start request is sent to an executor, it
retrieves the job’s executable code and runs a command to start the task. The
executor then monitors the process until it finishes or fails, at which point it
updates the task status metadata accordingly.
Many executors also work with the operating system to provide both security
and performance isolation. YARN and Kubernetes use Linux cgroups, for exam‐
ple. This prevents tasks from accessing data without permission or from nega‐
tively affecting the performance of other tasks on the node by using excessive
resources.
Resource manager
An orchestrator’s resource manager stores metadata about each node, including
available hardware (CPUs, GPUs, memory, disks, and so on), task statuses, net‐
work location, node status, and other relevant information. Thus, the manager
provides a global view of the cluster’s current state. The centralized nature of
the resource manager can lead to both scalability and availability bottlenecks.
YARN uses ZooKeeper, and Kubernetes uses etcd, to store cluster state (see
“Coordination Services” on page 437).
Scheduler
Orchestrators usually have a centralized scheduler subsystem, which receives
requests to start, stop, or check on the status of a job. For example, a scheduler
might receive a request to start a job with 10 tasks using a specific Docker image
on nodes that have a specific type of GPU. The scheduler uses the information
from the request and the state of the resource manager to determine which tasks
to run on which nodes. The task executors are then informed of their assigned
work and begin execution.
462 
| 
Chapter 11: Batch Processing


Though each orchestrator uses different terminology, you will find these components
in nearly all orchestration systems. 
Scheduling decisions sometimes require application-specific sched‐
ulers that can take into account particular requirements, such as
autoscaling read replicas when a certain query threshold is reached.
The centralized scheduler and application-specific schedulers work
together to determine how best to execute tasks. YARN refers to its
subschedulers as ApplicationMasters, while Kubernetes calls them
operators.
Resource allocation
Schedulers have a particularly challenging role in job orchestration. They must figure
out how to optimally allocate the cluster’s limited resources among jobs with compet‐
ing needs. Fundamentally, these decisions must balance fairness and efficiency.
Imagine a small cluster with five nodes that has a total of 160 CPU cores available.
The cluster’s scheduler receives two job requests, each wanting 100 cores to complete
its work. What’s the best way to schedule the workload?
• The scheduler could decide to run 80 tasks for each job, starting the remaining 20
•
tasks for each job as earlier tasks complete.
• The scheduler could run all of one job’s tasks, then begin running the second job’s
•
tasks only when 100 cores are available (a strategy known as gang scheduling).
• If the second job request arrives much later than the first, the scheduler has
•
incomplete information. It has to decide whether to allocate all 100 cores to the
first job, or hold some back in anticipation of a future job that might or might not
ever come.
This is a very simple example, but we already see many difficult trade-offs. In the
gang scheduling scenario, for example, if the scheduler reserves CPU cores until all
100 are available at the same time, nodes will sit idle. The cluster’s resource utilization
will drop, and a deadlock might occur if other jobs also attempt to reserve CPU cores.
On the other hand, if the scheduler simply waits for 100 cores to become available,
other jobs might grab the cores in the meantime. The cluster might not have 100
cores available for a very long time, which leads to starvation. Alternatively, the
scheduler could decide to preempt some of the first job’s tasks, killing them to make
room for the second job. However, task preemption decreases cluster efficiency as
well, since the killed tasks will need to be restarted later.
Now imagine a scheduler that must make allocation decisions for hundreds or even
millions of such job requests. Finding an optimal solution seems intractable. In fact,
Batch Processing in Distributed Systems 
| 
463


the problem is NP-hard, which means that it is prohibitively slow to calculate an
optimal solution for all but the smallest examples [14, 15].
In practice, schedulers therefore use heuristics to make nonoptimal but reasonable
decisions. Several algorithms are commonly used, including first-in first-out (FIFO),
dominant resource fairness (DRF), priority queues, capacity or quota-based schedul‐
ing, and various bin-packing algorithms. The details of such algorithms are beyond
the scope of this book, but they’re a fascinating area of research. 
Scheduling workflows
The Unix tools example in “Simple Log Analysis” on page 454 involved a chain of
several commands, connected by Unix pipes. The same pattern arises in distributed
batch processes: often the output from one job needs to become the input to one or
more other jobs, and each job may have several inputs that are produced by other
jobs. This is called a workflow or directed acyclic graph (DAG) of jobs.
In “Durable Execution and Workflows” on page 187 we saw work‐
flow engines that offer durable execution of a sequence of steps,
typically performing RPCs. In the context of batch processing,
“workflow” has a different meaning: it’s a sequence of batch pro‐
cesses, each taking input data and producing output data, but
normally not making RPCs to external services. Durable execution
engines typically process less data per request than their batch
processing counterparts, though the line is somewhat fuzzy.
A workflow of multiple jobs might be needed for several reasons:
• If the output of one job needs to become the input to several other jobs, which
•
are maintained by different teams, it’s best for the first job to write its output to
a location where all the other jobs can read it. Those consuming jobs can then be
scheduled to run every time that data has been updated or on another schedule.
• You might want to transfer data from one processing tool to another. For exam‐
•
ple, a Spark job might output its data to HDFS, then a Python script might trigger
a Trino SQL query (see “Cloud Data Warehouses” on page 135) that does further
processing on the HDFS files and outputs to S3.
• Some data pipelines internally require multiple processing stages. For example, if
•
one stage needs to shard the data by one key, and the next stage needs to shard by
a different key, the first stage can output data sharded in the way that is required
by the second stage.
In the Unix tools example, the pipe that connects the output of one command to the
input of another uses only a small in-memory buffer and doesn’t write the data into
a file. If that buffer fills up, the producing process needs to wait until the consuming
464 
| 
Chapter 11: Batch Processing


process has read some data from the buffer before it can output more—a form
of backpressure. Spark, Flink, and other batch execution engines support a similar
model where the output of one task is directly passed to another task (over the
network if the tasks are running on different machines).
However, it is more typical for one job in a workflow to write its output to a
distributed filesystem or object store and for the next job to read it from there. This
decouples the jobs from each other, allowing them to run at different times. If a job
has several inputs, a workflow scheduler typically waits until all the jobs that produce
its inputs have completed successfully before running the job that consumes those
inputs.
Schedulers found in orchestration frameworks such as YARN’s ResourceManager
or Spark’s built-in scheduler do not manage entire workflows; they do scheduling
on a per-job basis. To handle these dependencies between job executions, various
workflow schedulers have been developed, including Airflow, Dagster, and Prefect.
Workflow schedulers have management features that are useful when maintaining a
large collection of batch jobs. Workflows consisting of 50 to 100 jobs are common in
many data pipelines, and in a large organization many teams may be running jobs
or workflows that read one another’s output across many systems. Tool support is
important for managing such complex dataflows. 
Handling faults
Batch jobs often run for long periods of time. Long-running jobs with many parallel
tasks are likely to experience at least one task failure along the way. As discussed in
“Hardware and Software Faults” on page 44 and “Unreliable Networks” on page 347,
this could happen for reasons, including hardware faults (especially on commodity
hardware) or network interruptions.
Another reason a task might not finish running is that the scheduler may intentionally
preempt (kill) it. Preemption is particularly useful if you have multiple priority levels,
such as low-priority tasks that are cheaper to run and high-priority tasks that cost
more. Low-priority tasks can run whenever there is spare computing capacity, but they
run the risk of being preempted at any moment if a higher-priority task arrives. Such
cheaper, low-priority virtual machines are called spot instances on Amazon EC2, spot
virtual machines on Azure, and preemptible instances on Google Cloud [16].
Since batch processing is often used for jobs that are not time-sensitive, it is well
suited for using low-priority tasks and spot instances to reduce the cost of running
jobs. Essentially, those jobs can use spare computing resources that would otherwise
be idle, thereby increasing the utilization of the cluster. However, this also means that
those tasks are more likely to be killed by the scheduler, because preemptions occur
more frequently than hardware faults [17].
Batch Processing in Distributed Systems 
| 
465


Since batch jobs regenerate their output from scratch every time they are run, task
failures are easier to handle than in online systems. The system can delete the partial
output from the failed execution and schedule the task to run again on another
machine. It would be wasteful to rerun the entire job because of a single task failure,
though. MapReduce and its successors therefore keep the execution of parallel tasks
independent from each other, so that they can retry work at the granularity of an
individual task [3].
Fault tolerance is trickier when the output of one task becomes the input to another
as part of a workflow. MapReduce solves this by always writing such intermediate
data back to the distributed filesystem and waiting for the writing task to complete
successfully before allowing other tasks to read the data. This works, even in an
environment where preemption is common, but it means a lot of writes to the DFS,
which can be inefficient.
Spark keeps intermediate data in memory (“spilling” it to the local disk if it won’t fit),
and writes only the final result to the DFS. It also keeps track of how the intermediate
data was computed, allowing Spark to recompute it in case it is lost [18]. Flink uses a
different approach based on periodically checkpointing a snapshot of tasks [19]. We
will return to this topic in “Dataflow Engines” on page 468.
Batch Processing Models
We have seen how batch jobs are scheduled in a distributed environment. Let’s
now turn our attention to how batch processing frameworks process data. The two
most common models are MapReduce and dataflow engines. Although dataflow
engines have largely replaced MapReduce in practice, it is useful to understand how
MapReduce works since it influenced many modern batch processing frameworks.
MapReduce and dataflow engines have evolved to support multiple programming
models, including low-level programmatic APIs, relational query languages, and
DataFrame APIs. A variety of options enable application engineers, analytics engi‐
neers, business analysts, and even nontechnical employees to process company data
for various use cases, which we’ll discuss in “Batch Use Cases” on page 476.
MapReduce
The pattern of data processing in MapReduce is very similar to the web server log
analysis example in “Simple Log Analysis” on page 454:
1. Read a set of input files and break it into records. In the web server log example,
1.
each record is one line in the log (i.e., \n is the record separator). In Hadoop’s
MapReduce, the input file is stored in a distributed filesystem like HDFS or an
object store like S3. Various file formats are used, such as Apache Parquet (a
466 
| 
Chapter 11: Batch Processing


columnar format, see “Column-Oriented Storage” on page 136) or Apache Avro
(a row-based format, see “Avro” on page 172).
2. Call the mapper function to extract a key and value from each input record.
2.
In the Unix tools example, the mapper function is awk '{print $7}', which
extracts the URL ($7) as the key and leaves the value empty.
3. Sort all the key-value pairs by key. In the log example, this is done by the first
3.
sort command.
4. Call the reducer function to iterate over the sorted key-value pairs. If there are
4.
multiple occurrences of the same key, the sorting has made them adjacent in the
list, so it is easy to combine those values without having to keep a lot of state in
memory. In the Unix tools example, the reducer is implemented by the command
uniq -c, which counts the number of adjacent records with the same key.
Those four steps can be performed by one MapReduce job. Steps 2 (map) and 4
(reduce) are where you write your custom data processing code. Step 1 (breaking files
into records) is handled by the input format parser. Step 3, the sort step, is implicit
in MapReduce—you don’t have to write it, because the output from the mapper is
always sorted before it is given to the reducer. This sorting step is a foundational
batch processing algorithm, which we’ll revisit in “Shuffling Data” on page 469.
To create a MapReduce job, you need to implement two callback functions, the
mapper and reducer, which behave as follows:
Mapper
The mapper is called once for every input record, and its job is to extract the
key and value from the record. For each input, it may generate any number
of key-value pairs (including none). It does not keep any state from one input
record to the next, so each record is handled independently. There can be many
mappers, running in parallel on different parts of the input.
Reducer
The MapReduce framework takes the key-value pairs produced by the mappers,
collects all the values belonging to the same key, and calls the reducer with an
iterator over that collection of values. The reducer can produce output records
(such as the number of occurrences of the same URL). Reducers for different
keys can also be run in parallel.
In the web server log example, we had a second sort command in step 5, which
ranked URLs by number of requests. In MapReduce, if you need a second sorting
stage, you can implement it by writing a second MapReduce job and using the output
of the first job as input to the second job. Viewed like this, the role of the mapper is
to prepare the data by putting it into a form that is suitable for sorting, and the role of
the reducer is to process the data that has been sorted.
Batch Processing Models 
| 
467


MapReduce and Functional Programming
Though MapReduce is used for batch processing, the programming model comes
from functional programming. Lisp introduced map and reduce (or fold) as higher-
order functions on lists, and they have made their way into mainstream languages
such as Python, Rust, and Java.
Many common data processing operations, including those offered by SQL, can be
implemented on top of MapReduce. The functional programming principle of avoid‐
ing mutable state helps enable parallel execution. As every call to the mapper and
reducer depends only on the data that the MapReduce framework explicitly passes to
those functions, the framework is free to run independent calls in parallel on different
nodes. And if a task fails, the framework is free to call the mapper and reducer again
with the same input on another node.
Implementing a complex processing job using the raw MapReduce APIs is actually
quite laborious—for instance, any join algorithms used by the job would need to be
implemented from scratch [20]. MapReduce is also quite slow compared to more
modern batch processors. One reason is that its file-based I/O prevents job pipelin‐
ing (i.e., processing output data in a downstream job before the upstream job is
complete). 
Dataflow Engines
To fix some of MapReduce’s problems, several new execution engines for distributed
batch computations were developed, the most well-known of which are Spark [18, 21]
and Flink [19]. They are designed differently, but they have one thing in common:
they handle an entire workflow as one job, rather than breaking it into independent
subjobs.
Since they explicitly model the flow of data through several processing stages, these
systems are known as dataflow engines. Like MapReduce, they support a low-level API
that repeatedly calls a user-defined function to process one record at a time, but they
also offer higher-level operators such as join and group by. They parallelize work by
sharding inputs, and they copy the output of one task over the network to become the
input to another task. Unlike in MapReduce, operators need not take the strict roles of
alternating map and reduce, but instead can be assembled in more flexible ways.
These dataflow APIs generally use relational-style building blocks to express a com‐
putation: joining datasets on the value of a field; grouping tuples by key; filtering
by a condition; and aggregating tuples by counting, summing, or other functions.
Internally, these operations are implemented using the shuffle algorithms that we
discuss in the next section.
468 
| 
Chapter 11: Batch Processing


This style of processing engine is based on research systems like Dryad [22] and
Nephele [23], and it offers several advantages compared to the MapReduce model:
• Expensive work such as sorting needs to be performed only in places where it is
•
required, rather than always happening by default between every map and reduce
stage.
• When there are several operators in a row that don’t change the sharding of the
•
dataset (such as map or filter), they can be combined into a single task, reducing
data copying overheads.
• Because all joins and data dependencies in a workflow are explicitly declared,
•
the scheduler has an overview of what data is required where, so it can make
locality optimizations. For example, it can try to place the task that consumes
some data on the same machine as the task that produces it, so that the data can
be exchanged through a shared memory buffer rather than having to be copied
over the network.
• It is usually sufficient for intermediate state between operators to be kept in
•
memory or written to local disk, which requires less I/O than writing it to a
distributed filesystem or object store (where it must be replicated to several
machines and written to disk on each replica). MapReduce already uses this
optimization for mapper output, but dataflow engines generalize the idea to all
intermediate state.
• Operators can start executing as soon as their input is ready; there is no need to
•
wait for the entire preceding stage to finish before the next one starts.
• Existing processes can be reused to run new operators, reducing startup over‐
•
heads compared to MapReduce (which launches a new JVM for each task).
You can use dataflow engines to implement the same computations as MapReduce
workflows, and they usually execute significantly faster because of the optimizations
described here. 
Shuffling Data
As we’ve seen, both the Unix tools example at the beginning of the chapter and
MapReduce are based on sorting. Batch processors need to be able to sort datasets
petabytes in size, which are too large to fit on a single machine. They therefore
require a distributed sorting algorithm where both the input and the output are
sharded. Such an algorithm is called a shuffle.
Batch Processing Models 
| 
469


Shuffle is not random
The term shuffle is potentially confusing. When you shuffle a deck
of cards, you end up with a random order. In contrast, the shuffle
we’re talking about produces a sorted order, with no randomness.
Shuffling is a foundational algorithm for batch processors, where it is used for
joins and aggregations. MapReduce, Spark, Flink, Daft, Dataflow, and BigQuery [24]
all implement scalable and performant shuffle algorithms in order to handle large
datasets. We’ll use the shuffle in Hadoop MapReduce [25] for illustration purposes,
but the concepts in this section translate to other systems as well.
Figure 11-1 shows the dataflow in a MapReduce job. We assume that the input to the
job is sharded, and the shards are labeled m 1, m 2, and m 3. For example, each shard
may be a separate file on HDFS or a separate object in an object store, and all the
shards belonging to the same dataset are grouped into the same HDFS directory or
have the same key prefix in an object store bucket.
Figure 11-1. A MapReduce job with three mappers and three reducers
The framework starts a separate map task for each input shard. A task reads its
assigned file, passing one record at a time to the mapper callback. The reduce side of
the computation is also sharded. While the number of map tasks is determined by the
number of input shards, the number of reduce tasks is configured by the job’s author
(it can be different from the number of map tasks).
470 
| 
Chapter 11: Batch Processing


The output of the mapper consists of key-value pairs, and the framework needs to
ensure that if two mappers output the same key, those key-value pairs end up being
processed by the same reducer task. To achieve this, each mapper creates a separate
output file on its local disk for every reducer (e.g., the file m 1, r 2 in Figure 11-1
is the file created by mapper 1 containing the data destined for reducer 2). When
the mapper outputs a key-value pair, a hash of the key typically determines which
reducer file it is written to (similarly to the process described in “Sharding by Hash of
Key” on page 258).
While a mapper is writing these files, it also sorts the key-value pairs within each
file. This can be done using the techniques we saw in “Log-Structured Storage” on
page 118: batches of key-value pairs are first collected in a sorted data structure
in memory, then written out as sorted segment files, and smaller segment files are
progressively merged into larger ones.
After each mapper finishes, reducers connect to it and copy the appropriate file of
sorted key-value pairs to their local disk. Once the reduce task has its share of the
output from all the mappers, it merges these files together, preserving the sort order,
mergesort-style. Key-value pairs with the same key are now consecutive, even if they
came from different mappers. The reducer function is then called once per key, each
time with an iterator that returns all the values for that key.
Any records output by the reducer function are sequentially written to a file, with one
file per reduce task. These files (r 1, r 2, and r 3 in Figure 11-1) become the shards
of the job’s output dataset, and they are written back to the distributed filesystem or
object store.
Though MapReduce executes the shuffle step between its map and reduce steps,
modern dataflow engines and cloud data warehouses are more sophisticated. Systems
such as BigQuery have optimized their shuffle algorithms to keep data in memory
and to write data to external sorting services [24]. Such services speed up shuffling
and replicate shuffled data to provide resilience. 
Joins and Grouping
Let’s look at how sorted data simplifies distributed joins and aggregations. We’ll
continue with MapReduce for illustration purposes, though these concepts apply to
most batch processing systems.
A typical example of a join in a batch job is illustrated in Figure 11-2. On the left is
a log of events describing the things that logged-in users did on a website (known
as activity events or clickstream data), and on the right is a database of users. You
can think of this example as being part of a star schema (see “Stars and Snowflakes:
Schemas for Analytics” on page 77); the log of events is the fact table, and the user
database is one of the dimensions.
Batch Processing Models 
| 
471


Figure 11-2. A join between a log of user activity events and a database of user profiles
If you want to perform an analysis of the activity events that takes into account
information from the user database (e.g., find out whether certain pages are more
popular with younger or older users, using the date-of-birth field in the user profile),
you need to compute a join between these two tables. How would you compute that
join, assuming both tables are so large that they have to be sharded?
You can use the fact that in MapReduce, the shuffle brings together all the key-value
pairs with the same key to the same reducer, no matter which shard they were on
originally. Here, the user ID can serve as the key. You can therefore write a mapper
that goes over the user activity events and emits page view URLs keyed by user ID, as
illustrated in Figure 11-3. Another mapper goes over the user database row by row,
extracting the user ID as the key and the user’s date of birth as the value.
The shuffle then ensures that a reducer function can access a particular user’s date
of birth and all of that user’s page view events at the same time. The MapReduce job
can even arrange the records to be sorted such that the reducers always see the record
from the user database first, followed by the activity events in timestamp order. This
technique is known as a secondary sort [25].
The reducers can now perform the actual join logic easily. The first value is expected
to be the date of birth, which the reducer stores in a local variable. It then iterates
over the activity events with the same user ID, outputting each viewed URL along
with the viewer’s date of birth. Since a reducer processes all the records for a particu‐
lar user ID in one go, it needs to keep only one user record in memory at any one
time, and it never needs to make any requests over the network. This algorithm is
known as a sort-merge join, since mapper output is sorted by key and the reducers
then merge together the sorted lists of records from both sides of the join.
472 
| 
Chapter 11: Batch Processing


The next MapReduce job in the workflow can then calculate the distribution of
viewer ages for each URL. To do so, the job first shuffles the data using the URL as
the key. Once sorted, the reducers iterate over all the page views (with viewer birth
date) for a single URL, keeping a counter for the number of views by each age group
and incrementing the appropriate counter for each page view. This way, you can
implement a group by operation and aggregation. 
Figure 11-3. A sort-merge join on user ID; if the input datasets are sharded into multiple
files, each could be processed with multiple mappers in parallel
Query Languages
Over the years, execution engines for distributed batch processing have matured.
The infrastructure is now robust enough to store and process many petabytes of
data on clusters of over 10,000 machines. With the problem of physically operating
batch processes at such scale considered more or less solved, attention has turned to
improving the programming model.
MapReduce, dataflow engines, and cloud data warehouses have all embraced SQL
as the lingua franca for batch processing. It’s a natural fit, because legacy data ware‐
houses used SQL, data analytics and ETL tools already support it, and all developers
and analysts know it.
Besides requiring less code than handwritten MapReduce jobs, these query language
interfaces also allow interactive use, in which you write analytical queries and run
them from a terminal or GUI. This style of interactive querying is an efficient and
natural way for business analysts, product managers, sales and finance teams, and
others to explore data in a batch processing environment. SQL support has also made
distributed batch processing systems suitable for exploratory queries.
High-level query languages don’t just make the humans using the system more
productive; they also improve job execution efficiency at a machine level. As we
Batch Processing Models 
| 
473


saw in “Cloud Data Warehouses” on page 135, query engines are responsible for
converting SQL queries into batch jobs to be executed in a cluster. This translation
step from query to syntax tree to physical operators allows the engine to optimize
queries. Query engines such as Hive, Trino, Spark, and Flink have cost-based query
optimizers that can analyze the properties of join inputs and automatically decide
which algorithm would be most suitable for the task at hand. Optimizers might even
change the order of joins so that the amount of intermediate state is minimized [19,
26, 27, 28].
While SQL is the most popular general-purpose batch processing query language,
other languages remain in use for niche needs. For example, Apache Pig was a
language based on relational operators that allowed data pipelines to be specified
step by step, rather than as one big SQL query. DataFrames (discussed in the next
section) have similar characteristics, and Morel is a more modern language that
was influenced by Pig. Other users have adopted JSON query languages such as jq,
JMESPath, or JSONPath.
In “Graph-Like Data Models” on page 84 we discussed using graphs for modeling
data and using graph query languages to traverse the edges and vertices in a graph.
Many graph processing frameworks also support batch computation through query
languages such as Apache TinkerPop’s Gremlin. We will look at graph processing
scenarios in more detail in “Batch Use Cases” on page 476. 
Batch Processing and Cloud Data Warehouses Converge
Historically, data warehouses ran on specialized hardware appliances and supported
SQL-based analytical queries over relational data. Batch processing frameworks like
MapReduce set out to provide greater scalability and flexibility by supporting pro‐
cessing logic written in a general-purpose programming language, allowing reading
and writing of arbitrary data formats.
Over time, the two have become much more similar. Modern batch processing frame‐
works now support SQL as a language for writing batch jobs, and they achieve
good performance on relational queries by using columnar storage formats such as
Parquet and optimized query execution engines (see “Query Execution: Compilation
and Vectorization” on page 142). Meanwhile, data warehouses have grown more
scalable by moving to the cloud (see “Cloud Data Warehouses” on page 135) and
implementing many of the same scheduling, fault tolerance, and shuffling techniques
that distributed batch frameworks do. Many use distributed filesystems as well.
Just as batch processing systems adopted SQL as a processing model, cloud ware‐
houses have adopted alternative processing models as well. For example, BigQuery
offers a DataFrames library, and Snowflake’s Snowpark library integrates with Pandas.
Batch processing workflow orchestrators such as Airflow, Prefect, and Dagster also
integrate with cloud warehouses.
474 
| 
Chapter 11: Batch Processing


Not all batch jobs are easily expressed in SQL, though, including iterative graph
algorithms such as PageRank, complex ML tasks, and many other workflows. AI data
processing, which includes nonrelational and multimodal data such as images, video,
and audio, can also be difficult to express in SQL.
Cloud data warehouses struggle with certain workloads as well. Row-by-row com‐
putation is less efficient when using column-oriented storage formats; alternative
warehouse APIs or a batch processing system are preferable in such cases. Cloud data
warehouses also tend to be more expensive than other batch processing systems. It
can be more cost-efficient to run large jobs in batch processing systems such as Spark
or Flink instead.
Ultimately, the decision between processing data in batch systems or data warehouses
often comes down to factors such as cost, convenience, ease of implementation, and
availability. Most large enterprises have many data processing systems, which gives
them flexibility in this decision. Smaller companies often get by with just one.
DataFrames
Data scientists and statisticians are generally used to working with the DataFrame
data model found in R and Pandas (see “DataFrames, Matrices, and Arrays” on page
105). A DataFrame is similar to a table in a relational database: it is a collection of
rows, and all the values in the same column have the same type. Instead of writing
one big SQL query, users call functions corresponding to relational operators to
perform filters, joins, sorting, aggregations, and other operations.
Originally, DataFrame manipulation typically occurred locally, in memory. Conse‐
quently, DataFrames were limited to datasets that fit on a single machine. Data
scientists wanted to interact with the large datasets found in batch processing envi‐
ronments by using the DataFrame APIs they were used to, since SQL and MapReduce
are not well suited to their needs. Distributed data processing frameworks such
as Spark, Flink, and Daft have adopted DataFrame APIs to meet this need. Their
implementation behaves somewhat differently, though; local DataFrames are usually
indexed and ordered, while distributed DataFrames are generally not [29]. This can
lead to performance surprises when migrating to batch frameworks.
DataFrame APIs appear similar to dataflow APIs, but implementations vary. While
Pandas executes operations immediately when DataFrame methods are called, Spark
first translates all the DataFrame API calls into a query plan and runs query optimi‐
zation before executing the workflow on top of its distributed dataflow engine.
Frameworks such as Daft even support both client- and server-side computation.
Smaller, in-memory operations are executed on the client, and larger datasets are
processed on a server. Columnar storage formats such as Apache Arrow offer a
unified data model that both client- and server-side execution engines can share.
Batch Processing Models 
| 
475


Batch Use Cases
Now that we’ve seen how batch processing works, let’s see how it is applied to a range
of applications. Batch jobs are excellent for processing large datasets in bulk, but they
aren’t good for low-latency use cases. Consequently, you’ll find batch jobs wherever
there’s a lot of data and data freshness isn’t important. This might sound limiting,
but it turns out that a significant amount of data processing tasks fit this model. For
example:
• Accounting and inventory reconciliation, where companies verify that transac‐
•
tions line up with their bank accounts and inventory, are often performed as
batch jobs [30].
• In manufacturing, demand forecasting commonly runs as a periodic batch job
•
[31].
• Ecommerce, media, and social media companies train their recommendation
•
models with batch jobs [32, 33].
• Many financial systems are batch-based; for example, the US banking network
•
runs almost entirely on batch jobs [34].
In the following sections, we’ll discuss some of the batch processing use cases you’ll
find in nearly every industry.
Extract–Transform–Load
“Data Warehousing” on page 7 introduced ETL and ELT, where a data processing
pipeline extracts data from a production database, transforms it, and loads the results
into a downstream system (we’ll use “ETL” in this section to represent both ETL and
ELT workloads). Batch jobs are often used for such workloads, especially when the
downstream system is a data warehouse.
The parallel nature of batch jobs makes them a great fit for data transformation,
much of which involves “embarrassingly parallel” workloads. Filtering data, projec‐
ting fields, and many other common data warehouse transformations can all be done
in parallel.
Batch processing environments also come with robust workflow schedulers, which
make it easy to schedule, orchestrate, and debug ETL data pipeline jobs. When a
failure occurs, schedulers often retry jobs to mitigate transient issues that might
occur. A job that fails repeatedly will be marked as failed, which helps developers
easily see which job in their data pipeline stopped working. Schedulers like Airflow
even come with built-in source, sink, and query operators for MySQL, PostgreSQL,
Snowflake, Spark, Flink, and dozens of other popular systems. A tight integration
between schedulers and data processing systems simplifies data integration.
476 
| 
Chapter 11: Batch Processing


We’ve also seen that batch jobs are easy to troubleshoot and fix when things go awry.
This feature is invaluable when debugging data pipelines. Failed files can be easily
inspected to see what went wrong, and ETL batch jobs can be fixed and rerun. For
example, if an input file does not contain a field that a transformation batch job
intends to use, data engineers can easily spot that the field is missing and update the
transformation logic or the job that produced the input.
Data pipelines used to be managed by a single data engineering team, as it was
considered unfair to ask other teams working on product features to write and
manage complex batch data pipelines. Recently, improvements in batch processing
models and metadata management have made it much easier for engineers across an
organization to contribute to and manage their own data pipelines. Data mesh [35,
36], data contract [37], and data fabric [38] practices provide standards and tools to
help teams safely publish their data for consumption by anybody in the organization.
Data pipelines and analytical queries have begun to share not only processing models,
but execution engines as well. Many batch ETL jobs now run on the same systems
as the analytical queries that read their output. It is not uncommon to see both data
pipeline transformations and analytical queries run as SparkSQL, Trino, or DuckDB
queries. Such an architecture further blurs the line between application engineering,
data engineering, analytics engineering, and business analysis.
Analytics
In “Operational Versus Analytical Systems” on page 3, we saw that analytical queries
(OLAP) often scan over a large number of records, performing groupings and aggre‐
gations. It is possible to run such workloads in a batch processing system, alongside
other batch processing workloads. Analysts write SQL queries that execute atop a
query engine, which reads from and writes to a distributed filesystem or object store.
Table metadata such as table-to-file mappings, names, and types are managed with
table formats such as Apache Iceberg and catalogs such as Unity (see “Cloud Data
Warehouses” on page 135). This architecture is known as a data lakehouse [39].
As with ETL, improvements in SQL query interfaces mean many organizations now
use batch frameworks such as Spark for analytics. Such query patterns come in two
styles:
Pre-aggregation queries
Data is rolled up into OLAP cubes or data marts to speed up queries (see “Mate‐
rialized Views and Data Cubes” on page 143). Pre-aggregated data is queried
in the warehouse or pushed to purpose-built real-time OLAP systems such as
Apache Druid or Apache Pinot. Pre-aggregation normally takes place at a sched‐
uled interval. The workflow schedulers discussed in “Scheduling workflows” on
page 464 are used to manage these workloads.
Batch Use Cases 
| 
477


Ad hoc queries
Users run these to answer specific business questions, investigate user behavior,
debug operational issues, and much more. Response times are important for this
use case. Analysts run queries iteratively as they get responses and learn more
about the data they’re investigating. Batch processing frameworks with fast query
execution help reduce waiting times for analysts.
SQL support enables batch processing frameworks to integrate with spreadsheets and
data visualization tools such as Tableau, Power BI, Looker, and Apache Superset.
For example, Tableau offers SparkSQL and Presto connectors, while Apache Superset
supports Trino, Hive, Spark SQL, Presto, and many other systems that ultimately
execute batch jobs to query data. 
Machine Learning
Machine learning (ML) makes frequent use of batch processing. Data scientists, ML
engineers, and AI engineers use batch processing frameworks to investigate data
patterns, transform data, and train ML models. Common uses include the following:
Feature engineering
Raw data is filtered and transformed into data that models can be trained on.
Predictive models often need numeric data, so engineers must transform other
forms of data (such as text or discrete values) into the required format.
Model training
The training data is the input to the batch process, and the weights of the trained
model are the output.
Batch inference
A trained model can be used to make predictions in bulk if datasets are large
and real-time results are not required. This includes evaluating the model’s
predictions on a test dataset.
Batch processing frameworks provide tools explicitly for these use cases. For example,
Apache Spark’s MLlib and Apache Flink’s FlinkML come with a wide variety of
feature engineering tools, statistical functions, and classifiers.
ML applications such as recommendation engines and ranking systems also make
heavy use of graph processing (see “Graph-Like Data Models” on page 84). Many
graph algorithms are expressed by traversing one edge at a time, joining one vertex
with an adjacent vertex in order to propagate some information, and repeating until
a certain condition is met—for example, until there are no more edges to follow, or
until a metric converges.
The bulk synchronous parallel (BSP) model of computation [40] has become popular
for batch processing graphs; it’s implemented by Apache Giraph [20], Spark’s GraphX
478 
| 
Chapter 11: Batch Processing


API, and Flink’s Gelly API [41], among others. It is also known as the Pregel model, as
Google’s Pregel paper popularized this approach for processing graphs [42].
Batch processing is also an integral part of large language model data preparation and
training. Raw text input data such as the contents of websites typically resides in a
DFS or object store. This data must be preprocessed to make it suitable for training.
Preprocessing steps that are well suited for batch processing frameworks include the
following:
• Extracting plain text from HTML and fixing malformed text.
•
• Detecting and removing low-quality, irrelevant, and duplicate documents.
•
• Tokenizing text (splitting it into words) and converting it into embeddings, or
•
numeric representations of each word.
Batch processing frameworks such as Kubeflow, Flyte, and Ray are purpose-built
for such workloads. OpenAI uses Ray as part of its ChatGPT training process, for
example [43]. These frameworks have built-in integrations for LLM and AI libraries
such as PyTorch, TensorFlow, XGBoost, and many others. They also offer built-in
support for feature engineering, model training, batch inference, and fine-tuning
(adjusting a foundational model for specific use cases).
Finally, data scientists often experiment with data in interactive notebooks such
as Jupyter or Hex. Notebooks are made up of cells, which are small chunks of
Markdown, Python, or SQL. Cells are executed sequentially to produce spreadsheets,
graphs, or data. Many notebooks use batch processing via DataFrame APIs or query
such systems using SQL. 
Serving Derived Data
Batch jobs are often used to build precomputed or derived datasets such as product
recommendations, user-facing reports, and features for ML models. These datasets
are typically served from a production database, key-value store, or search engine.
Regardless of the system used, the precomputed data needs to make its way from the
batch processor’s distributed filesystem or object store back into the database that’s
serving live traffic.
You might be tempted to use the client library for your favorite database directly within
a batch job and write directly to the database server, one record at a time. This will
work (assuming your firewall rules allow direct access from your batch processing
environment to your production databases), but it is a bad idea for several reasons:
• Making a network request for every single record is orders of magnitude slower
•
than the normal throughput of a batch task. Even if the client library supports
batching, performance is likely to be poor.
Batch Use Cases 
| 
479


• Batch processing frameworks often run many tasks in parallel. If all the tasks
•
concurrently write to the same output database at the rate expected of a batch
process, that database can easily be overwhelmed, and its performance for quer‐
ies is likely to suffer. This can in turn cause operational problems in other parts of
the system [44].
• Normally, batch jobs provide a clean all-or-nothing guarantee for job output. If
•
a job succeeds, the result is the output of running every task exactly once, even
if some tasks failed and had to be retried along the way; if the entire job fails,
no output is produced. However, writing to an external system from inside a job
produces externally visible side effects that cannot be hidden in this way. Thus,
you have to worry about the results from partially completed jobs being visible
to other systems. If a task fails and is restarted, it may duplicate output from the
failed execution.
A better solution is to have batch jobs push precomputed datasets to streams such
as Kafka topics, which we discuss further in Chapter 12. Search engines like Elas‐
ticsearch, real-time OLAP systems like Apache Pinot and Apache Druid, derived
datastores like Venice [45], and cloud data warehouses like ClickHouse all have the
built-in ability to ingest data from Kafka into their systems. Pushing data through a
streaming system fixes a few of the aforementioned problems:
• Streaming systems are optimized for sequential writes, which makes them better
•
suited for the bulk write workload of a batch job.
• Streaming systems can act as a buffer between the batch job and the production
•
databases. Downstream systems can throttle their read rate to ensure they can
continue to comfortably serve production traffic.
• The output of a single batch job can be consumed by multiple downstream
•
systems.
• Streaming systems can serve as a security boundary between batch processing
•
environments and production networks. They can be deployed in a so-called
demilitarized zone (DMZ) network that sits between the batch processing net‐
work and the production network.
One issue streaming doesn’t inherently solve is that of the all-or-nothing guarantee.
To make this work, upon completion batch jobs must send a notification to down‐
stream systems that their job is done and the data can now be served. Consumers of
the stream need to be able to keep the data they receive invisible to queries, like an
uncommitted transaction with read-committed isolation (see “Read Committed” on
page 290), until they are notified that the job is complete.
480 
| 
Chapter 11: Batch Processing


Another pattern that is more common when bootstrapping databases is to build a
brand-new database inside the batch job and bulk-load those files directly into the
database from a distributed filesystem, object store, or local filesystem. Many data
systems offer bulk import tools, such as TiDB’s Lightning and Apache Pinot’s Hadoop
import jobs. RocksDB also offers an API to bulk-import Sorted String Table (SST)
files from batch jobs.
Building databases in batch and bulk-importing the data is very fast and makes it
easier for systems to atomically switch between dataset versions. On the other hand, it
can be challenging to incrementally update datasets from batch jobs that build brand-
new databases. It’s common to take a hybrid approach when both bootstrapping and
incremental loads are needed. Venice, for example, supports hybrid stores that allow
for batch row-based updates and full dataset swaps. 
Summary
In this chapter we explored the design and implementation of batch processing
systems. We began with the classic Unix toolchain (awk, sort, uniq, etc.) to illustrate
fundamental batch processing primitives such as sorting and counting.
We then scaled up to distributed batch processing systems. Batch frameworks process
immutable, bounded input datasets to produce output data, allowing reruns and
debugging without side effects. This processing involves three main components:
an orchestration layer that determines where and when jobs run, a storage layer to
persist data, and a computation layer that processes the actual data.
We looked at how distributed filesystems and object stores manage large files through
block-based replication, caching, and metadata services, and how modern batch
frameworks interact with these systems via pluggable APIs. We also discussed how
job orchestrators schedule tasks, allocate resources, and handle faults in large clusters,
and we compared them with workflow orchestrators that manage the lifecycle of a
collection of jobs that run in a dependency graph.
We surveyed batch processing models, starting with MapReduce and its canonical
map and reduce functions. Next, we turned to dataflow engines like Spark and Flink,
which offer simpler-to-use dataflow APIs and better performance. To understand
how batch jobs scale, we covered the shuffle algorithm, a foundational operation that
enables grouping, joining, and aggregation.
We saw that as batch systems matured, focus shifted to usability. Support was added
for high-level query languages like SQL and DataFrame APIs, making batch jobs
more accessible and easier to optimize. The batch framework takes jobs written in
these languages and automatically determines how to execute them efficiently on a
cluster of machines.
Summary 
| 
481


We finished the chapter with a survey of common batch processing use cases, includ‐
ing the following:
• ETL pipelines, which extract, transform, and load data between systems using
•
scheduled workflows
• Analytics, where batch jobs support both pre-aggregated and ad hoc queries
•
• Machine learning, where batch jobs are used to prepare and process large train‐
•
ing datasets
• Populating production-facing systems from batch outputs, often via streams or
•
bulk-loading tools, in order to serve the derived data to users
In the next chapter we will turn to stream processing, in which the input is unboun‐
ded—that is, a job’s inputs are never-ending streams of data. This means jobs are
never complete because more work may be coming in at any time. We shall see that
stream and batch processing are similar in some respects, but the assumption of
unbounded streams also has a significant impact on how we build systems. 
References
[1] Nathan Marz. “How to Beat the CAP Theorem.” nathanmarz.com, October 2011.
Archived at perma.cc/4BS9-R9A4
[2] Molly Bartlett Dishman and Martin Fowler. “Agile Architecture.” At O’Reilly
Software Architecture Conference, March 2015.
[3] Jeffrey Dean and Sanjay Ghemawat. “MapReduce: Simplified Data Processing
on Large Clusters.” At 6th USENIX Symposium on Operating System Design and
Implementation (OSDI), December 2004.
[4] Shivnath Babu and Herodotos Herodotou. “Massively Parallel Databases and
MapReduce Systems.” Foundations and Trends in Databases, volume 5, issue 1, pages
1–104, November 2013. doi:10.1561/1900000036
[5] David J. DeWitt and Michael Stonebraker. “MapReduce: A Major Step Back‐
wards.” Originally published at databasecolumn.vertica.com, January 2008. Archived
at perma.cc/U8PA-K48V
[6] Henry Robinson. “The Elephant Was a Trojan Horse: On the Death of Map-
Reduce at Google.” the-paper-trail.org, June 2014. Archived at perma.cc/9FEM-X787
[7] Urs Hölzle. “R.I.P. MapReduce. After having served us well since 2003, today
we removed the remaining internal codebase for good.” x.com, September 2019.
Archived at perma.cc/B34T-LLY7
482 
| 
Chapter 11: Batch Processing


[8] Adam Drake. “Command-Line Tools Can Be 235x Faster than Your Hadoop
Cluster.” aadrake.com, January 2014. Archived at perma.cc/87SP-ZMCY
[9] "sort: Sort Text Files.” GNU Coreutils 9.7 Documentation, Free Software Founda‐
tion, Inc., 2025. Archived at perma.cc/68KN-E8TL
[10] Michael Ovsiannikov, Silvius Rus, Damian Reeves, Paul Sutter, Sriram Rao, and
Jim Kelly. “The Quantcast File System.” Proceedings of the VLDB Endowment, volume
6, issue 11, pages 1092–1101, August 2013. doi:10.14778/2536222.2536234
[11] Andrew Wang, Zhe Zhang, Kai Zheng, Uma Maheswara G., and Vinayakumar
B. “Introduction to HDFS Erasure Coding in Apache Hadoop.” blog.cloudera.com,
September 2015. Archived at archive.org
[12] Andy Warfield. “Building and Operating a Pretty Big Storage System Called S3.”
allthingsdistributed.com, July 2023. Archived at perma.cc/7LPK-TP7V
[13] Vinod Kumar Vavilapalli, Arun C. Murthy, Chris Douglas, Sharad Agarwal,
Mahadev Konar, Robert Evans, Thomas Graves, Jason Lowe, Hitesh Shah, Sid‐
dharth Seth, Bikas Saha, Carlo Curino, Owen O’Malley, Sanjay Radia, Benjamin
Reed, and Eric Baldeschwieler. “Apache Hadoop YARN: Yet Another Resource
Negotiator.” At 4th Annual Symposium on Cloud Computing (SoCC), October 2013.
doi:10.1145/2523616.2523633
[14] Richard M. Karp. “Reducibility Among Combinatorial Problems.” Complex‐
ity of Computer Computations. The IBM Research Symposia Series. Springer, 1972.
doi:10.1007/978-1-4684-2001-2_9
[15] J. D. Ullman. “NP-Complete Scheduling Problems.” Journal of Computer
and System Sciences, volume 10, issue 3, pages 384–393, June 1975. doi:10.1016/
S0022-0000(75)80008-0
[16] Gilad David Maayan. “The Complete Guide to Spot Instances on AWS, Azure
and GCP.” datacenterdynamics.com, March 2021. Archived at archive.org
[17] Abhishek Verma, Luis Pedrosa, Madhukar Korupolu, David Oppenheimer,
Eric Tune, and John Wilkes. “Large-Scale Cluster Management at Google with
Borg.” At 10th European Conference on Computer Systems (EuroSys), April 2015.
doi:10.1145/2741948.2741964
[18] Matei Zaharia, Mosharaf Chowdhury, Tathagata Das, Ankur Dave, Justin Ma,
Murphy McCauley, Michael J. Franklin, Scott Shenker, and Ion Stoica. “Resilient Dis‐
tributed Datasets: A Fault-Tolerant Abstraction for In-Memory Cluster Computing.”
At 9th USENIX Symposium on Networked Systems Design and Implementation (NSDI),
April 2012.
Summary 
| 
483


[19] Paris Carbone, Stephan Ewen, Seif Haridi, Asterios Katsifodimos, Volker Markl,
and Kostas Tzoumas. “Apache Flink: Stream and Batch Processing in a Single Engine.”
Bulletin of the IEEE Computer Society Technical Committee on Data Engineering,
volume 38, issue 4, pages 28–38, December 2015. Archived at perma.cc/G3N3-BKX5
[20] Mark Grover, Ted Malaska, Jonathan Seidman, and Gwen Shapira. Hadoop
Application Architectures. O’Reilly Media, 2015. ISBN: 9781491900048
[21] Jules S. Damji, Brooke Wenig, Tathagata Das, and Denny Lee. Learning Spark,
2nd edition. O’Reilly Media, 2020. ISBN: 9781492050049
[22] Michael Isard, Mihai Budiu, Yuan Yu, Andrew Birrell, and Dennis Fetterly.
“Dryad: Distributed Data-Parallel Programs from Sequential Building Blocks.”
At 2nd European Conference on Computer Systems (EuroSys), March 2007.
doi:10.1145/1272996.1273005
[23] Daniel Warneke and Odej Kao. “Nephele: Efficient Parallel Data Processing in
the Cloud.” At 2nd Workshop on Many-Task Computing on Grids and Supercomputers
(MTAGS), November 2009. doi:10.1145/1646468.1646476
[24] Hossein Ahmadi. “In-Memory Query Execution in Google BigQuery.” cloud.goo‐
gle.com, August 2016. Archived at perma.cc/DGG2-FL9W
[25] Tom White. Hadoop: The Definitive Guide, 4th edition. O’Reilly Media, 2015.
ISBN: 9781491901632
[26] Fabian Hüske. “Peeking into Apache Flink’s Engine Room.” flink.apache.org,
March 2015. Archived at perma.cc/44BW-ALJX
[27] Mostafa Mokhtar. “Hive 0.14 Cost Based Optimizer (CBO) Technical Overview.”
hortonworks.com, March 2015. Archived at archive.org
[28] Michael Armbrust, Reynold S. Xin, Cheng Lian, Yin Huai, Davies Liu,
Joseph K. Bradley, Xiangrui Meng, Tomer Kaftan, Michael J. Franklin, Ali
Ghodsi, and Matei Zaharia. “Spark SQL: Relational Data Processing in Spark.”
At ACM International Conference on Management of Data (SIGMOD), June 2015.
doi:10.1145/2723372.2742797
[29] Kaya Kupferschmidt. “Spark vs. Pandas, Part 2—Spark.” towardsdatascience.com,
October 2020. Archived at perma.cc/5BRK-G4N5
[30] Ammar Chalifah. “Tracking Payments at Scale.” bolt.eu.com, June 2025. Archived
at perma.cc/Q4KX-8K3J
[31] Nafi Ahmet Turgut, Hamza Akyıldız, Hasan Burak Yel, Mehmet İkbal Özmen,
Mutlu Polatcan, Pinar Baki, and Esra Kayabali. “Demand Forecasting at Getir Built
with Amazon Forecast.” aws.amazon.com, May 2023. Archived at perma.cc/H3H6-
GNL7
484 
| 
Chapter 11: Batch Processing


[32] Jason (Siyu) Zhu. “Enhancing Homepage Feed Relevance by Harnessing the
Power of Large Corpus Sparse ID Embeddings.” linkedin.com, August 2023. Archived
at archive.org
[33] Avery Ching, Sital Kedia, and Shuojie Wang. “Apache Spark @Scale: A 60 TB+
Production Use Case.” engineering.fb.com, August 2016. Archived at perma.cc/F7R5-
YFAV
[34] Edward Kim. “How ACH Works: A Developer Perspective—Part 1.” engineer‐
ing.gusto.com, April 2014. Archived at perma.cc/F67P-VBLK
[35] Zhamak Dehghani. “How to Move Beyond a Monolithic Data Lake to a Dis‐
tributed Data Mesh.” martinfowler.com, May 2019. Archived at perma.cc/LN2L-L4VC
[36] Chris Riccomini. “What the Heck Is a Data Mesh?!” cnr.sh, June 2021. Archived
at perma.cc/NEJ2-BAX3
[37] Chad Sanderson, Mark Freeman, and B. E. Schmidt. Data Contracts. O’Reilly
Media, 2025. ISBN: 9781098157623
[38] Daniel Abadi. “Data Fabric vs. Data Mesh: What’s the Difference?” starburst.io,
November 2021. Archived at perma.cc/RSK3-HXDK
[39] Michael Armbrust, Ali Ghodsi, Reynold Xin, and Matei Zaharia. “Lakehouse:
A New Generation of Open Platforms That Unify Data Warehousing and Advanced
Analytics.” At 11th Annual Conference on Innovative Data Systems Research (CIDR),
January 2021. Archived at perma.cc/7C6D-T9NR
[40] Leslie G. Valiant. “A Bridging Model for Parallel Computation.” Com‐
munications of the ACM, volume 33, issue 8, pages 103–111, August 1990.
doi:10.1145/79173.79181
[41] Stephan Ewen, Kostas Tzoumas, Moritz Kaufmann, and Volker Markl. “Spinning
Fast Iterative Data Flows.” Proceedings of the VLDB Endowment, volume 5, issue 11,
pages 1268–1279, July 2012. doi:10.14778/2350229.2350245
[42] Grzegorz Malewicz, Matthew H. Austern, Aart J. C. Bik, James C. Dehnert,
Ilan Horn, Naty Leiser, and Grzegorz Czajkowski. “Pregel: A System for Large-Scale
Graph Processing.” At ACM International Conference on Management of Data (SIG‐
MOD), June 2010. doi:10.1145/1807167.1807184
[43] Richard MacManus. “OpenAI Chats About Scaling LLMs at Anyscale’s Ray
Summit.” thenewstack.io, September 2023. Archived at perma.cc/YJD6-KUXU
[44] Jay Kreps. “Why Local State Is a Fundamental Primitive in Stream Processing.”
oreilly.com, July 2014. Archived at perma.cc/P8HU-R5LA
[45] Félix GV. “Open Sourcing Venice—LinkedIn’s Derived Data Platform.”
linkedin.com, September 2022. Archived at archive.org
Summary 
| 
485