پردازش دسته‌ایBatch Processing

This chapter's architecture reads as three layers — and Diagram 2 draws exactly them: storage (a distributed filesystem or object store), orchestration (scheduler, resource manager and workflow scheduler) and compute (isolated tasks each reading one input shard and writing one output shard). Processing models sit on these layers: MapReduce, dataflow engines (like Spark and Flink), query languages like SQL and DataFrame APIs. And the applications group into four: ETL, analytics, machine learning and serving derived data.

Most things built so far were “ask, get an answer”: the browser wants a page, a remote service calls an API, the database, cache and search answer. These are online systems and their core metric is response time. But sometimes you hold a computation that fits no interactive request: training a model, transforming a huge data volume, or analyzing a large dataset. These are batch jobs, and the systems running them are offline systems.

A batch job's contract is simple: read read-only input, build output from scratch. Unlike a read/write transaction it never mutates data; output is derived from input. Dislike the output? Delete it, fix the job logic, rerun. This “immutable input + no side effects” buys several big wins:

• Recovering from code bugs: if you ship a bug and corrupt the output, just revert to the previous code version and rerun the job — or simpler: you kept the previous output under another path and switch back to it. Most object stores and open table formats support this, calling it time travel. Transactional databases usually lack this property: if buggy code writes bad data, reverting the code fixes nothing. This recovery-from-code-errors ability is called human fault tolerance.
• Faster development: when a mistake means no “irreversible damage”, teams ship features faster. This is the same “minimize irreversibility” principle that serves agile development well.
• Reusing one file set: the same input can feed different jobs, including monitoring jobs that compare a job's output with the previous run and measure the drift.
• Resource efficiency: batch frameworks consume resources efficiently; batch work on online systems (like an OLTP database or app server) is possible but far costlier in resources.

Some batch systems handle a task failure by “killing and rerunning the whole job”; others have fault tolerance: the job still completes though several nodes died. The online/batch boundary isn't always sharp — a long database query looks a lot like batch processing — but this batch model is what turns it into a building block of dependable, scalable, maintainable systems; e.g. in data integration and ETL. (Batch's alternative is stream processing: the job never ends after the input runs out and processes changes as they happen — Chapter 12.)

Modern batch processing is deeply shaped by MapReduce — the algorithm Google published in 2004, later implemented in Hadoop, CouchDB and MongoDB. MapReduce is a fairly low-level programming model; less capable than the parallel execution engines of analytical databases. Freshly arrived, it was a leap in processable scale on commodity hardware, but today it is largely obsolete and no longer used at Google.

Today's batch processing usually runs on Spark or Flink or data-warehouse query engines. Like MapReduce they rely on sharding (Chapter 7) and parallel execution, but with far more sophisticated caching and execution strategies. As these systems matured, operational concerns were largely solved and focus moved to usability: new processing models like dataflow APIs, query languages and DataFrame APIs are now supported everywhere. Orchestration tooling matured too: Hadoop-centric schedulers like Oozie and Azkaban gave way to general solutions like Airflow, Dagster and Prefect supporting a wide range of frameworks and cloud warehouses.

And of course the cloud: batch storage is moving from distributed filesystems like HDFS, GlusterFS and CephFS to object stores like S3; and scalable cloud data warehouses like BigQuery and Snowflake blurred the line between warehouse and batch processing. Result: “convergence” — seen precisely in §4.6.

The best intuition for batch processing is what already works on one machine: a chain of Unix tools joined by pipes. Later we generalize the same example to distributed.

Suppose a web server appending one line to a log file per request. A standard NGINX access line looks like this:

(This is really one line; split for readability.) The log format is defined as:

So this line says: at the given date the server got a request for /css/typography.css from IP 216.58.210.78; the user wasn't authenticated (so $remote_user is a dash); the response was 200 with 3,377 bytes. Now let's find the “top five pages” with basic tools:

• cat /var/log/nginx/access.log — reads the log file. (Honestly cat is unneeded here; the file could go to awk directly. But the linear-pipeline style reads clearer.)
• awk '{print $7}' — splits each line on whitespace and prints only the seventh field, the requested URL; in the sample line, /css/typography.css.
• sort — sorts the URL list alphabetically. Why: if a URL was requested n times, the sorted output holds that same URL n times in a row.
• uniq -c — drops adjacent duplicate lines; the -c flag adds a counter: one line with its count per distinct URL.
• sort -r -n — sorts a second time, now by the leading number (-n), the request count, in reverse (-r) so the biggest comes first.
• head -n 5 — keeps only the first five lines and discards the rest.

That pipeline's output looks like this:

The command line may look cryptic, but it is remarkably powerful: it chews through gigabytes of logs in seconds and is easy to change. To drop CSS files from the report, change awk's argument to $7 !~ /\.css$/ {print $7}; and for busiest client IPs instead of pages, just put {print $1} — the first field is the IP.

The same job could be done with a simple program instead of the Unix chain. E.g. in Python:

• counts = defaultdict(int) — a hash table keeping one counter per URL; each counter starts at zero.
• line.split()[6] — takes the requested URL from each line; the seventh whitespace-separated field (index 6, since Python arrays count from zero).
• counts[url] += 1 — increments that line's URL counter.
• sorted(... reverse=True)[:5] — sorts the hash table's contents by counter (descending) and takes the top five.
• print(...) — prints those five.

This program is less compact than the Unix chain but readable, and the choice between them is partly taste. But one big execution-flow difference shows on large files:

• In-memory aggregation (the Python script): the URL hash table lives in memory. The job's memory footprint (the memory needing random access — called the working set) depends only on the distinct URL count: a URL seen a million times still takes one URL + one counter. When this working set is small (e.g. most websites fit in 1 GB), this method is excellent even on a laptop.
• Sorting (the pipeline): when the working set outgrows memory, sorting wins because it uses disk efficiently — exactly the “log-structured storage” principle (Chapter 4): data chunks are sorted in memory and written as segment files, then several small sorted segments merge into a bigger sorted file. Mergesort's sequential access pattern performs far better on disk (and SSD) than random access.
• And the practical point: GNU Coreutils sort itself handles bigger-than-memory datasets by spilling to disk and parallelizes sorting across CPU cores. So the same simple command line scales to large datasets without running out of memory; the bottleneck is likely the input file's read rate.

Unix tools' limit is running on one machine. Datasets that fit neither in memory nor on local disk are the problem — and that is where distributed batch frameworks enter. The same Unix example had three cooperating parts: storage devices reachable through the OS filesystem interface; a scheduler deciding when each process runs and how much CPU it gets; and a set of Unix programs whose stdin and stdout are piped together. The same three parts exist in distributed frameworks — the key analogy of the next section.

As seen at the previous section's end, a distributed batch framework can be viewed as a distributed operating system: it has a filesystem, a job scheduler, and programs passing data through the filesystem or other channels. Three main layers:

• Storage: blocks, metadata and a swappable access protocol — the filesystem role.
• Orchestration: what runs where with how much memory and CPU — the kernel role.
• Compute: isolated tasks each reading one input shard and writing one output shard — the processes role.

Your own OS filesystem has layers: at the bottom the block device driver talks to the disk directly, enabling raw block reads/writes; above it the page cache keeps recently-touched blocks in memory; then the filesystem layer (like ext4 or XFS) breaks big files into blocks and tracks metadata like inodes and directories; and finally the OS exposes filesystems to programs through one shared API called VFS.

Distributed filesystems (DFS) work the same way: files break into blocks spread across machines; but DFS blocks are usually far bigger: HDFS defaults to 128MB, JuiceFS and many object stores to 4MB — vs ext4's 4,096 bytes. Bigger blocks mean less metadata (a big deal on petabyte datasets) and cheaper seeks relative to reads. One subtle difference: physical devices cannot write “part of a block” so the OS must write whole blocks, but since a DFS is implemented on the nodes' OS filesystems it has no such requirement: a 900MB file with 128MB blocks means seven 128MB blocks and one 4MB block.

DFS blocks are read with network requests to the machine holding the block. Each machine runs a daemon offering a block read/write API as files on its local filesystem: Hadoop calls them DataNodes, GlusterFS calls them glusterfsd processes. The distributed equivalent of the page cache exists too: since blocks sit as files on data nodes, every read passes through that node's OS page cache; some DFSs add more cache layers (like client-side caches and local disk in JuiceFS).

Metadata has its own problem: ext4/XFS track free space, file-block placement, directory structure and permissions; a DFS likewise needs a way to track file placement across machines. Hadoop has a service called NameNode; 3FS (DeepSeek's project) has a metadata service persisting its data in a KV store like FoundationDB. And the VFS role is played by the DFS protocol: any DFS implementing that protocol is usable. The S3 API became ubiquitous exactly this way: MinIO, Cloudflare R2, Tigris, Backblaze B2 and many more implement it. Some DFSs are fully POSIX-compatible and attach to VFS via FUSE or NFS — NFS being the most famous distributed-filesystem protocol.

One final architectural dimension: DFSs are built on the shared-nothing principle — vs the shared-disk approach of NAS and SAN, centralized appliances (sometimes with special hardware and Fibre Channel). Shared-nothing wants only commodity computers + an ordinary datacenter network. Commodity hardware is cheaper but fails more often; so blocks are replicated across machines (like Chapter 6) or protected with erasure coding like Reed–Solomon, cheaper in storage than full copies — the RAID idea, but over the datacenter network and without special hardware.

Object storage services like Amazon S3, Google Cloud Storage, Azure Blob Storage and OpenStack Swift became the popular DFS replacement for batch jobs. The boundary between the two is blurring: FUSE drivers let you see an object store as a filesystem, and some DFS implementations like JuiceFS and Ceph offer both APIs (object and filesystem). But their APIs, performance and consistency guarantees differ a lot — adopt carefully and verify they really behave as you expect.

• URL structure: each object has an address like s3://my-photo-bucket/2025/04/01/birthday.png. The host part (my-photo-bucket) is the bucket name (globally unique); the rest is the object's key, unique within the bucket.
• Get/put and immutability: reading is get, writing is put. Unlike files, an object is immutable after writing: changing it means rewriting it whole — like a key-value store. (Azure Blob and S3 Express One Zone have appends; most others don't.) No file handles like fopen/fseek exist.
• Prefixes instead of directories: paths look directory-like but object stores have no directory concept; slashes are part of the key. Listing happens by prefix and differs from ls in two ways: it is recursive like ls -R (returning all objects below), and empty directories are impossible — deleting every object under a path makes the path vanish from listings; the common convention is creating a zero-byte object as the empty-directory stand-in.
• No links or locks; non-atomic rename: DFSs often have hard links, symbolic links, file locks and atomic rename; object stores don't. Rename means copy-to-new-key then delete-old-key; and renaming a “directory” means renaming every object one by one, since the directory name is part of the key.
• Difference from KV stores (Chapter 4): KV stores optimize for small values (kilobytes) and frequent low-latency reads/writes; DFS and object stores for big objects (megabytes to gigabytes) with bigger, rarer reads. Object stores are shifting this boundary though (e.g. S3 Express One Zone with single-digit-ms latency).
• Locality: a DFS like HDFS lets a task run on a machine holding a copy of the file — so the file never crosses the network, and if the task code is smaller than the data it reads, bandwidth is saved. Object stores usually keep storage and compute apart: more bandwidth is consumed (acceptable on today's fast networks), but CPU and memory scale independently of storage.

The same OS analogy holds in orchestration: running a Unix job means someone must actually launch the awk, sort and uniq processes, carry data from one's output to the next one's input, allocate memory and guarantee fair execution. On one machine that is the kernel's job; in a distributed setting it is the job orchestrator's. A job-run request carries metadata: task count, each task's memory/CPU/disk, the job id, access credentials, parameters (like input and output), needed hardware (like GPUs) and the runnable code's location.

• Task executor: a daemon like YARN's NodeManager or Kubernetes' kubelet sits on each node: it runs tasks, sends heartbeats to prove liveness, and tracks task state and the node's resource allocation. When a task-start request arrives it fetches the job code and fires the run command. Many executors also give OS-level security and performance isolation (e.g. cgroups in YARN and Kubernetes) so no task can reach data unauthorized or slow its node's other tasks by overconsuming.
• Resource manager: keeps each node's metadata: available hardware (CPU, GPU, memory, disk), task states, network position and node health — a global view of the cluster's current state. Its centrality can become a scalability and availability bottleneck; YARN writes state to ZooKeeper and Kubernetes to etcd.
• Scheduler: usually a centralized subsystem taking job start/stop/state requests — e.g. “a job with 10 tasks and such Docker image on nodes with such GPU type”. The scheduler combines the request with the resource manager's state to decide where each task runs and notifies the executors. Some decisions need application-specific schedulers (e.g. scaling read replicas when query counts pass a threshold); the centralized scheduler and these work together — YARN calls them ApplicationMasters, Kubernetes calls them operators.

The scheduler's job is genuinely hard: it must distribute the cluster's limited resources “optimally” among competing jobs, and these decisions inherently balance fairness against efficiency. See the book's puzzle: a 5-node cluster with 160 CPU cores total, and two job requests each wanting 100 cores. What to do?

• Run 80 tasks of each job and start each one's remaining 20 as earlier tasks free up.
• Run all of one's tasks and start the second only when 100 cores are free — a strategy called gang scheduling.
• If the second request arrives much later, the scheduler holds incomplete information: it must guess whether to give all 100 cores to the first job or hold some back for a job that may never come.

Even this simple example overflows with hard trade-offs. In gang scheduling, waiting for 100 simultaneously-free cores idles nodes, drops utilization, and if other jobs also want reservations a deadlock may strike. Waiting aimlessly for 100 free cores, other jobs may snatch cores and the cluster may lack 100 free cores for very long — starvation. The third way is preemption: kill some of the first job's tasks to make room for the second; but killed tasks must rerun later and cluster efficiency drops again.

In the Unix example each command's output reached the next command; distributed processing has the same pattern: one job's output becomes one or more other jobs' input. This is called a workflow, or a DAG (directed acyclic graph) of jobs.

Three common reasons jobs chain:

• If one job's output feeds several jobs maintained by different teams, the first job should write its output where all can read it; consuming jobs run whenever data refreshes (or on their own schedule).
• Sometimes you want to move data between processing tools: e.g. a Spark job outputs on HDFS, a Python script triggers a Trino query over the same files, and the result lands on S3.
• Sometimes the pipeline itself needs stages: if one stage shards data by one key and the next needs sharding by another, the first stage prepares output exactly as the second wants it.

The interesting engineering note: in the Unix example the pipe is just a small in-memory buffer with nothing filed; when the buffer fills, the producer waits for the consumer to read — a kind of backpressure. Spark and Flink have a similar model: one task's output goes straight to the next task (over the network if on different machines). But more common is separating jobs by writing intermediate output to DFS/object store: this gives time-decoupling, and when a job has several inputs the workflow scheduler waits for all producing jobs to succeed. And note: YARN's or Spark's internal schedulers work job-by-job and never grasp the whole workflow; for dependencies, workflow schedulers like Airflow, Dagster and Prefect were built. 50-to-100-job workflows are ordinary in data pipelines, and in a big organization many teams run jobs on each other's outputs — so real management tooling is needed.

Batch jobs run long; a long job with many parallel tasks will almost surely meet at least one task failure on its way — hardware faults (especially on commodity hardware) or network cuts. But a more common cause exists: the scheduler deliberately preempts (kills) the task. This pays off with priority levels: cheap low-priority tasks run whenever spare capacity exists, at the risk of being killed the moment a higher-priority job arrives. These cheap VMs are called spot instances on EC2, spot VMs on Azure and preemptible instances on Google Cloud. Since batch jobs are rarely urgent, using this cheap capacity is economical (and raises cluster use); but it means preemption happens far more than hardware failure.

Because batch jobs rebuild output from scratch every time, task-failure handling is simpler than in online systems: delete the failed run's partial output and reschedule the task on another machine. But rerunning the whole job for one broken task is wasteful; so MapReduce and its successors keep parallel tasks' execution independent, enabling single-task-level retries.

Things get hard when one task's output feeds another's input (in a workflow). Three famous strategies for intermediate data:

• MapReduce: writes all intermediate data to DFS and waits for the task's write to complete; works even in preemption-heavy settings, but means enormous DFS writes and low performance.
• Spark: keeps intermediate data in memory (spilling to local disk when needed) and writes only the final result to DFS; knowing how data was computed (lineage), it recomputes on loss.
• Flink: the checkpoint approach: periodically snapshots task state to durable storage instead of materializing every intermediate record.

So far we saw how jobs are scheduled and run; now how frameworks process data. Two common models: MapReduce and dataflow engines. Though dataflow has in practice replaced MapReduce, understanding MapReduce is still required — it shaped many modern frameworks. Both models opened over time to multiple programming models: low-level programming APIs, relational query languages and DataFrame APIs.

MapReduce's processing pattern closely resembles Section 2's log analysis — four steps:

1. Read input and break into records: in the log example each record is one line (record separator \n). In Hadoop, input sits in HDFS or an object store like S3, with various file formats: e.g. Apache Parquet (columnar) or Apache Avro (row-based).
2. Call the mapper to extract key and value from each input record. In our example the mapper is awk '{print $7}', taking the URL as key with an empty value.
3. Sort all (key, value) pairs by key. In the example, the first sort command.
4. Call the reducer on the sorted pairs. Since same-key records became adjacent through sorting, combining their values needs little in-memory state. In the example the reducer is uniq -c, counting adjacent same-key records.

These four steps fit in one MapReduce job. Steps 2 (map) and 4 (reduce) are where you write custom code; step 1 is solved by the input parser; and step 3, sorting, is implicit in MapReduce — you never write it, because mapper output is always sorted before reaching the reducer.

• Mapper: called exactly once per input record, extracting key and value; it may emit any number of pairs (even zero) per input. It keeps no state from the previous record — each record is processed independently — and many mappers can run in parallel over different input parts.
• Reducer: the framework collects the mappers' pairs, brings all values of one key together and calls the reducer with an iterator over that set. The reducer may build output records (like each URL's occurrence count). Reducers of different keys run in parallel too.

In the log example the second sort was for ranking. In MapReduce, needing a second sort stage means writing a second MapReduce job with the first job's output as its input. Roles from this angle: the mapper prepares data for sorting, and the reducer processes sorted data.

To fix MapReduce's problems, several new execution engines were built, most famously Spark and Flink. Their designs differ but share one thing: they see the whole workflow as one job, not a set of independent subjobs. Since they explicitly model data flow across processing stages, they are called dataflow engines. Like MapReduce they have a low-level API calling user functions record by record, but also higher-level operators like join and group by; they parallelize by sharding input and carry each task's output over the network to the next task's input. Unlike MapReduce, operators needn't alternate in a rigid map/reduce pattern and compose more flexibly. The style's roots are the Dryad and Nephele research systems.

Six advantages of this model over MapReduce:

• Expensive work like sorting happens only where truly needed, not always between every map and reduce.
• Consecutive operators that don't change the dataset's sharding (like map or filter) can fuse into one task, cutting data-copy costs.
• Since all joins and data dependencies are declared explicitly, the scheduler sees the whole picture and can optimize locality: e.g. placing the consuming task on the producer's machine so data passes through shared memory buffers, not the network.
• Intermediate state between operators can usually live in memory or local disk — far less I/O than writing to DFS/object store (which must be written several times over several disks). MapReduce had this optimization for mapper output; dataflow generalizes it to all intermediate state.
• Operators can start as soon as their input is ready; no need for the whole previous stage to finish.
• Existing processes are reused for running new operators, cutting startup costs (MapReduce boots a fresh JVM per task).

Dataflow can express the same computations as MapReduce workflows, and usually runs them observably faster thanks to these optimizations.

As seen, both the Unix example and MapReduce are built on sorting. Batch processors must sort petabyte datasets — data fitting no single machine — so they need a distributed sort whose input and output are both sharded. That algorithm is called shuffle.

We saw the Hadoop mechanism (Diagram 6, Figure 11-1): each mapper writes one sorted file per reducer, routing by hash(key), and the reducer merges its share mergesort-style. But dataflow engines and cloud warehouses are more advanced: systems like BigQuery optimized the shuffle algorithm to hold data in memory, or hand sorting to a dedicated external sort service; such services speed shuffle and replicate shuffled data so losing one node never stops the whole job.

Now see how sorted data simplifies distributed joins and aggregation. The classic example: on the left, the logged-in users' event log (called activity events or clickstream) and on the right, the user database. Think of this as part of a star schema: the event log is the fact table, the user database one dimension. To analyze with user info included (e.g. whether certain pages are more popular among younger or older users — via the birth-date field), you must join the two tables. Assuming both are big enough to shard, how?

The answer is that same shuffle property: it delivers every same-key pair, from whichever shard, to one reducer. So take user id as key: one mapper scans the events and publishes visited URLs keyed by user_id; another mapper scans the user database row by row, extracting user id as key and birth date as value. Shuffle guarantees the reducer holds one user's birth date and all their events at once. MapReduce can even order records so the reducer always sees the user-database record first, then events in timestamp order — a technique called secondary sort.

Now the join logic is simple: the first value must be the birth date, which the reducer keeps in a local variable; then it walks the same user_id's events, emitting each visited URL alongside the viewer's birth date. Since the reducer processes one user_id's records together, it needs only one user record in memory at a time and never a network request. This algorithm is the sort-merge join: mapper output is key-sorted and the reducer merges both join sides' sorted lists.

And the workflow's next job can compute viewer-age distribution per URL: shuffle data keyed by URL, and reducers walk all of one URL's visits, keeping one counter per age group and incrementing on each visit — i.e. group by and aggregation.

Distributed execution engines matured: infrastructure is now strong enough to store and process tens of petabytes on 10,000+-machine clusters. With operations solved, focus moved to the programming model. MapReduce, dataflow engines and cloud warehouses all adopted SQL as batch processing's lingua franca — a natural pick: old warehouses had SQL, analytics and ETL tools already supported it, and every developer and analyst knows it.

• Interactivity: instead of long hand-written MapReduce jobs, write a short query and run it from a terminal or GUI. This interactive query style is the natural way for business analysts, product managers, sales and finance teams to explore data; and SQL made batch systems fit exploratory queries too.
• Machine optimization: a high-level language doesn't just make humans more productive; it executes more efficiently too. The query engine turns SQL into a syntax tree and then physical operators, and that translation is where the cost-based optimizer lives: Hive, Trino, Spark and Flink can analyze join inputs' properties and pick the fittest algorithm — even reorder joins to minimize intermediate state.

But SQL isn't the only general language, and other languages hold niches: Apache Pig was a relational-operator language defining the pipeline step by step instead of one big SQL query — Morel is a more modern Pig-inspired language; DataFrames (next section) share a similar trait. Some turned to JSON query languages: jq, JMESPath, JSONPath. And for graph data (the data-models chapter), graph-processing frameworks support languages like Gremlin (Apache TinkerPop) for batch computation.

Historically two separate streams existed: data warehouses on dedicated hardware running analytical SQL over relational data; and batch frameworks like MapReduce arriving for bigger scale and flexibility — with logic in general-purpose languages and reads/writes of arbitrary formats. Over time the two grew very alike:

• Modern batch frameworks support SQL and, with columnar formats like Parquet and optimized execution engines, perform well on relational queries too.
• Warehouses became scalable in the cloud, implementing the same scheduling, fault-tolerance and shuffle techniques batch frameworks had; many also use distributed filesystems.
• And vice versa: cloud warehouses adopted alternative processing models — BigQuery has a DataFrames library and Snowpark in Snowflake integrates with Pandas. Orchestration tools like Airflow, Prefect and Dagster also plug into cloud warehouses.

But not every batch job expresses easily in SQL: iterative graph algorithms like PageRank, complex ML work, and non-relational, multimodal AI data processing (images, video, audio). Cloud warehouses also misfit some loads: row-by-row computation over columnar formats is inefficient (there another warehouse API or a batch system does better), and they are usually pricier — running Spark or Flink can be cheaper for big jobs. The final selection metric: cost, comfort, implementation simplicity and skill availability. Big companies usually run several processing systems (and the flexibility to choose); smaller ones often get by with one.

Data scientists and statisticians usually live in the DataFrame data model of R and Pandas: something like a relational table — a set of rows with same-typed values per column — except instead of one big SQL query you call relational-operator functions: filter, join, sort, aggregation and the rest.

DataFrame manipulation started local and in-memory, limited to datasets fitting one machine. Data scientists wanted the same familiar API on batch environments' big datasets (SQL and MapReduce didn't fit their needs); so distributed frameworks like Spark, Flink and Daft adopted DataFrame APIs. But the implementation behaves a bit differently, with a few important semantic differences:

• Local DataFrames are usually indexed and ordered; distributed DataFrames usually aren't. Migrating to batch frameworks can therefore bring “performance surprises” — code runs, but results or ordering aren't what you expected.
• Eager vs lazy: Pandas executes the moment a DataFrame method is called; Spark first translates all API calls into a query plan, runs optimization, then executes the workflow on its distributed dataflow engine.
• Client/server: frameworks like Daft have both: small in-memory ops run on the client, big datasets on the server. Columnar formats like Apache Arrow give a shared data model both the client-side and server-side engines can share.

We walked the previous seven subsections one by one: from MapReduce and shuffle to dataflow engines, query languages and DataFrames. Worth seeing them all once more in one picture — not as “one request's path” but as a class diagram: which concept subclasses which, what depends on what, and which ideas are shared across models.

The map's core message: what the market calls “batch processors” isn't several unrelated technologies; it is a stack. Atop the stack sit the programming models (Unix pipes, MapReduce, dataflow, SQL, DataFrames) which only set how computation is expressed. Mid-stack sit the execution engine and Shuffle. And below, Storage and the Orchestrator, separate from the programming model and joined by swappable protocols — the same idea as the “distributed operating system”.

Now that we know the mechanism, see where it is used in practice. Batch jobs excel at crunching massive datasets but misfit low latency; so you find them wherever data is plenty and data freshness doesn't matter. It may sound limited, but a large share of data processing fits this model: accounting and warehouse reconciliation, demand forecasting in manufacturing, recommender-model training in e-commerce and media, and many financial systems (the US banking network is almost entirely batch).

ETL (Extract–Transform–Load) and ELT mean a data-processing pipeline extracting data from the operational database, transforming it and loading the result into the downstream system (usually a warehouse). Batch jobs are perfect for this workload, especially with a warehouse downstream.

• Embarrassingly parallel: batch jobs' parallel nature suits data transformation perfectly; most common warehouse transforms — filtering, projecting fields — parallelize easily.
• Workflow schedulers: batch environments have strong workflow schedulers simplifying pipeline scheduling and debugging. On error, schedulers often retry the job to absorb transient faults; a persistently failing job is marked failed so developers easily see which pipeline job stalled. Airflow even ships ready operators for MySQL, PostgreSQL, Snowflake, Spark, Flink and dozens of other systems.
• Debuggability: when things break, batch jobs are easy to diagnose and repair: broken files can be inspected to learn what happened; e.g. if input lacks a transform's needed field, the data engineer plainly sees the missing field and fixes the transform logic or the producing job.
• Distributed ownership: data pipelines used to be run by one data-engineering team; improved processing models and metadata management now let each team's engineers write their own pipelines. Practices like data mesh, data contracts and data fabric give standards and tooling for teams to publish their data securely for the rest of the organization.
• Shared execution engine: data pipelines and analytical queries now share not just the processing model but the execution engine too — seeing pipeline transforms and analytical queries as SparkSQL, Trino or DuckDB queries is ordinary; and this further blurs the line between application, data, analytics engineering and business analysis.

Analytical (OLAP) queries often scan huge record counts and group and aggregate. Such a workload can run on the batch system alongside the other batch jobs: the analyst writes SQL, the query engine executes, reading and writing DFS/object store. Table metadata (mapping tables to files, names, types) is managed with a table format like Apache Iceberg and a catalog like Unity — this architecture is called data lakehouse.

Analytical queries on batch frameworks usually take two styles:

• Pre-aggregation: data is summarized in an OLAP cube or data mart to speed queries. Pre-aggregated data is either queried in the warehouse or pushed to dedicated real-time OLAP systems like Apache Druid or Apache Pinot. Pre-aggregation usually runs on a fixed cadence, managed by those same workflow schedulers.
• Ad hoc: run by a user answering a specific business question, studying user behavior or debugging operations. Here response time matters: the analyst queries iteratively, learning more with each answer; batch frameworks cut waiting time by executing queries fast.

And SQL support means integration with spreadsheets and visualization tools: Tableau, Power BI, Looker and Apache Superset — e.g. Superset supports Trino, Hive, Spark SQL and Presto, which are batch jobs underneath.

ML uses batch processing heavily: data scientists and ML/AI engineers discovering data patterns, transforming data and training models. Three common uses:

• Feature engineering: raw data is filtered and transformed into trainable data; predictive models usually want numeric data, so other forms (text or discrete values) must convert.
• Model training: training data is the batch input; the trained model's weights are the output.
• Batch inference: when the dataset is big and no real-time result is needed, the trained model predicts in bulk — including evaluating predictions on the test dataset.

Frameworks ship dedicated tooling for these: MLlib in Spark and FlinkML in Flink with feature-engineering tools, statistics functions and classifiers. Uses like recommenders and ranking also lean hard on graph processing: many graph algorithms express as traversing one edge per step — joining a vertex to neighbors to propagate information, iterating to a condition (e.g. until no edges remain or a metric converges). The bulk synchronous parallel (BSP) model grew popular for graph processing — in Apache Giraph, Spark's GraphX API and Flink's Gelly — also known as the Pregel model.

Batch processing is inseparable from preparing and training large language models (LLMs) too. Raw text data (like website contents) usually sits in DFS/object store and must be preprocessed: extracting plain text from HTML and fixing broken text, detecting and dropping low-quality/irrelevant/duplicate documents, and tokenizing (splitting into words) plus embedding. Frameworks like Kubeflow, Flyte and Ray were built for exactly these loads, with ready PyTorch, TensorFlow and XGBoost integration (plus built-in feature engineering, training, batch inference and fine-tuning). And finally, data scientists often experiment in interactive notebooks like Jupyter or Hex: small Markdown/Python/SQL cells executed in order with tables and charts as output — and many notebooks use the DataFrame API or SQL as their batch processing.

Batch jobs often build precomputed or derived datasets: product recommendations, user-facing reports and ML-model features. These are usually served from a production database, key-value store or search engine; so data must travel from the batch processor's DFS/object store to the database answering live traffic.

The better way: batch jobs push ready datasets to streams like Kafka topics. Search engines like Elasticsearch, real-time OLAP systems like Pinot and Druid, derived data stores like Venice and cloud warehouses like ClickHouse can all ingest from Kafka. Pushing through a streaming system fixes several problems:

• Streaming systems optimize for sequential writes — fitting a batch job's bulk write load.
• They can act as a buffer between the batch job and production databases: downstream systems throttle their read rate so they can comfortably serve production traffic at the same time.
• One batch job's output can feed several downstream systems.
• Streaming systems can be a security boundary between the batch environment and the production network: deployed in a network called a DMZ (demilitarized zone) between the two.

But one thing streams don't inherently fix: the all-or-nothing guarantee. For that, the job must announce completion to downstream systems at its end, and the data is servable only then. Stream consumers must keep received data hidden from queries until further notice — exactly like an uncommitted transaction under read committed isolation (Chapter 7) — until the job-done news arrives.

Another pattern, especially when bootstrapping databases: building a brand-new database inside the batch job and bulk-loading files straight from DFS/object store/local disk into the database. Bulk-import tools abound: TiDB Lightning, Hadoop import jobs in Pinot, and the bulk-import API for SST files in RocksDB. Building the database in batch and bulk-loading is very fast and simplifies the atomic dataset-version switch; but incremental updates on these freshly-built databases are challenging. When both bootstrapping and gradual loading are needed, a hybrid approach is common — e.g. Venice with its “hybrid store” has both batched row updates and full dataset replacement.

With all that said, the final decision states simply: by which path should a job's output reach the production serving system? The decision tree below orders the questions you must ask yourself one by one in practice — and its first branch shows that very “direct-write temptation” as a dead end.

• ❌ “Batch means updating data in batches.” → ✅ No; job input is immutable and output is built from scratch; the job never touches existing data.
• ❌ “Shuffle means scattering data randomly.” → ✅ Shuffle yields sorted output; it underpins joins and grouping, not randomness.
• ❌ “For each input change, we only process the changed part.” → ✅ In the classic batch model, even one changed byte means reprocessing the whole input.
• ❌ “When a task fails, we rerun the job from the start.” → ✅ Retries are task-level; the whole job fails only when tasks remain with budgets spent.
• ❌ “Preempted means failed.” → ✅ Preemption is no error: the task requeues penalty-free; only Failed eats retry budget.
• ❌ “Job output is visible while shards are being written.” → ✅ Output stays hidden until the job-done announcement (all-or-nothing, read-committed reveal).
• ❌ “For publication we write record by record straight to the production DB.” → ✅ Slow, drowning, and all-or-nothing-breaking; push to a stream or bulk-load + atomic switch instead.
• ❌ “An object store is a filesystem too; it has rename and locks.” → ✅ Rename is non-atomic (copy+delete), no links or locks exist, and empty directories are impossible.
• ❌ “The scheduler can compute the optimal resource allocation.” → ✅ The problem is NP-hard; only heuristics (FIFO, DRF, quotas, bin-packing) exist.
• ❌ “Workflow here is Chapter 5's durable execution.” → ✅ Here it means a DAG of batch jobs with input/output data, usually no RPC.
• ❌ “A distributed DataFrame behaves exactly like Pandas.” → ✅ It lacks the local version's indexing and ordering and is lazy (query plan) — migration surprises.
• ❌ “HDFS and S3 are architecturally one thing.” → ✅ HDFS gives near-data execution (locality); object stores separate compute and storage with different guarantees.

1. The batch-job definition: immutable input, from-scratch derived output, throughput over response time.
2. The model's three wins (time travel, human fault tolerance, faster development) and two prices (output only at job end, full-input reprocessing).
3. The log pipeline explained: awk, sort, uniq -c, sort -r -n, head — and why sort is needed.
4. Sort vs in-memory hash aggregation: working set, spills, mergesort and sequential access.
5. The platform's three layers (storage/orchestration/compute) and the OS analogy.
6. Object store's three key filesystem differences: immutability, prefix/recursive listing, no atomic rename or locks.
7. The allocation puzzle: fairness/efficiency, gang scheduling, preemption, NP-hardness and heuristics.
8. The three intermediate-data strategies and why task-level retry is safe.
9. The shuffle mechanism (hash(key), per-reducer files like m1r2, reduce-side mergesort) and sort-merge join with secondary sort.
10. Dataflow's six wins over MapReduce and the all-or-nothing publication story (Kafka push + job-done announcement).