MapReduce & Large-Scale Data

Before MapReduce, scaling a computation to terabytes of data meant either buying a very expensive single machine or writing deeply custom distributed code — both unsatisfying options. The breakthrough was recognizing a structural pattern in a large class of problems.

Data parallelism is the observation that many computations apply the same operation independently to many data items. If you want to count every word in the entire English-language Wikipedia, there is no reason word 1 through word 1,000,000 needs to wait for word 1,000,001 through word 2,000,000. The counts are independent. You can process each chunk simultaneously on a different machine, then add up the partial counts at the end.

This is different from task parallelism, where distinct different tasks run concurrently. Data parallelism scales with the size of your data, not the number of distinct operations you have. This is exactly what you need for big data.

The key insight is that these problems are embarrassingly parallel — a term of art meaning almost no inter-task communication is required during the parallel phase. Most of the data processing problems that actually matter at scale (log analysis, text indexing, ETL pipelines, training data preprocessing) fit this mold. MapReduce formalized this observation into a programming model.

# The embarrassingly parallel pattern (Python illustration)
# Sequential: processes 1 billion records one at a time
def word_count_sequential(records):
    counts = {}
    for record in records:           # bottleneck: 1 machine
        for word in record.split():
            counts[word] = counts.get(word, 0) + 1
    return counts

# Data-parallel: process N chunks simultaneously on N machines
def word_count_chunk(chunk):
    counts = {}
    for record in chunk:
        for word in record.split():
            counts[word] = counts.get(word, 0) + 1
    return counts                    # partial result; merge later

# Each machine runs word_count_chunk on its local data;
# a final step merges the partial dicts by summing values.

MapReduce, as described in the 2004 Google paper by Dean and Ghemawat, distills data-parallel computation into two user-defined functions inspired by their namesakes in functional programming.

Map takes a key-value pair and emits zero or more intermediate key-value pairs:

map(k1, v1) → list of (k2, v2)

Reduce takes an intermediate key and the list of all values for that key, and emits zero or more output values:

reduce(k2, list(v2)) → list(v3)

The framework handles everything in between: splitting the input, distributing map tasks to workers, shuffling (routing each (k2, v2) pair to the machine responsible for reducing k2), sorting by key so the reducer sees all values for a key together, and writing output.

Word Count — the canonical example. Input: a large text file. The mapper tokenizes each line and emits (word, 1) for every word. After the shuffle, every reducer receives one word and a list of 1s. The reducer sums them.

The model's power comes from what it constrains. Map has no side effects and sees no other record's data. Reduce sees all values for a key and produces a final answer. These constraints make it trivial for the framework to retry failed tasks, run speculative duplicate tasks, and guarantee correct output even when machines die.

# MapReduce word count in pseudo-Python
# (In real Hadoop Streaming this would be actual stdin/stdout)

def mapper(line):
    for word in line.strip().lower().split():
        yield (word, 1)              # emit (key, value)

def reducer(word, values):
    yield (word, sum(values))        # aggregate values for one key

# Framework orchestration (conceptual):
# 1. Split input into M chunks, one per mapper
# 2. Run mapper on each chunk in parallel
# 3. SHUFFLE: sort all (word, 1) pairs; route word→reducer R
# 4. Run reducer on each unique word in parallel
# 5. Write R output files

# Example flow for the string "the cat sat on the mat":
# Map:     (the,1),(cat,1),(sat,1),(on,1),(the,1),(mat,1)
# Shuffle: [(cat,[1]),(mat,[1]),(on,[1]),(sat,[1]),(the,[1,1])]
# Reduce:  (cat,1),(mat,1),(on,1),(sat,1),(the,2)

A cluster of 1,000 commodity machines has a daily hardware failure rate of roughly 1%. Run a job that takes an hour on 1,000 machines and you expect a failure. Any viable distributed execution model must handle this gracefully.

MapReduce's answer is elegant: make tasks stateless and deterministic, so failed tasks can simply be re-run.

The master node (called the JobTracker in Hadoop 1) monitors all worker nodes via heartbeat. If a worker does not respond within a timeout window, the master assumes it is dead and reschedules all in-progress or completed map tasks on that worker to another machine. (Map tasks are rescheduled even when complete, because their output on the dead machine's local disk is now inaccessible. Reduce tasks that have already fetched their map output do not need to be rerun.)

Speculative execution is a related mechanism: if a task is running significantly slower than similar tasks (a straggler), the framework launches a duplicate copy of the same task on a different machine, and whichever finishes first is used. This is safe precisely because map and reduce functions are required to be deterministic and have no side effects — two identical executions produce identical output, so it does not matter which one wins.

The master's own fault tolerance is handled by checkpointing its state (the map of task IDs to worker assignments and statuses) to durable storage periodically. A master failure requires re-running from the last checkpoint.

This design trades off redundant computation (re-running tasks) for simplicity in the programming model. The programmer does not write recovery logic. The constraint that functions be pure (deterministic, no side effects) is the price, and it is almost always a fair one.

# Pseudocode: Master's task scheduler loop
class Master:
    def __init__(self, input_splits, num_reducers):
        self.map_tasks = {i: IDLE for i in input_splits}
        self.reduce_tasks = {r: IDLE for r in range(num_reducers)}
        self.worker_heartbeat = {}   # worker_id → last seen time

    def assign_task(self, worker_id):
        self.worker_heartbeat[worker_id] = now()
        # Find any IDLE map task; assign it
        for task, status in self.map_tasks.items():
            if status == IDLE:
                self.map_tasks[task] = IN_PROGRESS(worker_id)
                return MapTask(task)
        # All maps done? Assign reduce tasks.
        ...

    def detect_failures(self):
        for worker, last_seen in self.worker_heartbeat.items():
            if now() - last_seen > TIMEOUT:
                # Reset all tasks assigned to this worker back to IDLE
                for task, status in self.map_tasks.items():
                    if status == IN_PROGRESS(worker):
                        self.map_tasks[task] = IDLE  # will be retried

MapReduce processes data. But where does the data live? You cannot fit a petabyte on a single disk. The answer is a distributed file system — a filesystem whose storage is spread across hundreds or thousands of machines, but that presents a single namespace to clients.

Google's GFS (2003 paper by Ghemawat, Gobioff, and Leung) and its open-source sibling HDFS (Hadoop Distributed File System) share the same key design decisions, each driven by workload reality:

Large block sizes (64 MB–256 MB). GFS uses 64 MB blocks; HDFS defaults to 128 MB. This is 100x larger than a typical filesystem block. Why? Because reading a 100 GB file sequentially has negligible metadata overhead with large blocks, and metadata costs dominate with small blocks at petabyte scale. The trade-off is wasted space for small files — HDFS is a poor fit for millions of tiny files.

Replication factor of 3. Every block is stored on 3 DataNodes (replica placement: one on the local rack, one on a different node in the same rack, one on a different rack). This tolerates two simultaneous node failures and allows reads to be served from the nearest replica.

Single master (NameNode in HDFS). The master stores all metadata (file namespace, file-to-block mapping, block-to-DataNode mapping) in memory for fast access. DataNodes store actual block data. This master bottleneck is a known limitation; HDFS HA uses a standby NameNode to mitigate it.

Write-once, append-only semantics. Files are written once and read many times (or appended to). Random writes are not supported. This dramatically simplifies consistency: there is no need to handle concurrent writers modifying the same byte range. The GFS paper is explicit: 'We have relaxed GFS's consistency model to vastly simplify the file system without imposing an onerous burden on the applications.'

Data locality. The MapReduce scheduler knows which blocks are on which machines. Whenever possible, it schedules a map task on the same machine (or at least the same rack) that holds the input block. This data locality eliminates most network I/O for the map phase — the dominant optimization in MapReduce performance.

# HDFS block placement illustration (pseudocode)
class HDFSNameNode:
    def write_file(self, filename, data):
        blocks = self.split_into_blocks(data, block_size=128*MB)
        block_locations = []
        for block in blocks:
            # Choose 3 DataNodes for replication
            # Rule: 1 local rack, 1 same rack, 1 remote rack
            dn1 = self.pick_local_node()          # writer's rack
            dn2 = self.pick_same_rack_node(dn1)   # same rack, different node
            dn3 = self.pick_remote_rack_node(dn1) # different rack entirely
            pipeline = [dn1, dn2, dn3]
            self.write_block_pipeline(block, pipeline)
            block_locations.append((block.id, pipeline))
        self.namespace[filename] = block_locations

# The scheduler exploits this metadata:
# for each map task whose input is block B,
# prefer to run the task on dn1, dn2, or dn3 (data-local).

The shuffle is the part of MapReduce that programmers often underestimate, and it is almost always the bottleneck. After all mappers finish, every (k2, v2) pair must be routed to the correct reducer. The reducer responsible for key k2 is determined by a partitioning function, typically hash(k2) mod R where R is the number of reducers.

This means every mapper must potentially send data to every reducer — an all-to-all communication pattern over the network. If you have M mappers and R reducers, and each map output has size S, then up to M × S bytes must cross the network during the shuffle. At petabyte scale, this is hundreds of gigabytes of network traffic, and the network is the slowest component in the cluster.

Combiner optimization. One powerful mitigation is the combiner (also called a local reducer). After the mapper produces its output but before the shuffle, the combiner runs on the local machine and partially aggregates values for the same key. For word count, the combiner sums up all (the, 1) pairs on one machine into a single (the, 47) before sending it over the network. This can reduce shuffle traffic by orders of magnitude for count-like operations.

Sort-merge. The shuffle also sorts all intermediate (key, value) pairs by key before handing them to reducers. Reducers do not receive a random stream of values; they receive a sorted iterator. This is why the standard Hadoop reduce signature receives key, Iterable[values] — you can process billions of values without materializing them all in memory, as long as they arrive sorted.

Skew is the enemy. If your key distribution is skewed (e.g., a 'hot' key like the word 'the' appears a billion times), one reducer will receive far more data than others. This is a data skew problem and it will stall your job waiting for the slow reducer. Solutions include custom partitioners, salting keys (adding a random prefix to distribute hot keys across multiple reducers and merging in a second pass), and pre-aggregation with combiners.

# Combiner: partial aggregation before shuffle
# Reducer function and combiner function are identical for
# associative, commutative operations like sum/max/count.

def mapper_with_combiner(line, combine_buffer):
    for word in line.strip().split():
        combine_buffer[word] = combine_buffer.get(word, 0) + 1
    # Emit combine_buffer periodically to limit memory usage:
    if memory_pressure():
        for word, count in combine_buffer.items():
            yield (word, count)   # goes into local sort buffer
        combine_buffer.clear()

# Without combiner: 1 billion (the, 1) records cross the network
# With combiner:    1 record (the, 47000) per mapper crosses network
# Speedup factor: proportional to the frequency of the hot key

Apache Hadoop (2006, by Doug Cutting and Mike Cafarella at Yahoo, open-sourcing the ideas from Google's papers) made MapReduce and the GFS-style distributed filesystem available to everyone. At its core, Hadoop 1.x had two components: HDFS (the distributed file system) and MapReduce (the compute layer with its own resource manager called JobTracker/TaskTracker).

Hadoop 2.x introduced YARN (Yet Another Resource Negotiator), which separated cluster resource management from the MapReduce programming model. YARN is a general-purpose cluster scheduler: it manages CPU and memory across the cluster and allows multiple execution engines (MapReduce, Spark, Tez, Flink) to run simultaneously on the same cluster. This was a significant architectural improvement — before YARN, the entire cluster was locked into the MapReduce execution model.

The Hadoop ecosystem grew into a rich stack:

• Hive: SQL-like query language (HiveQL) that compiles to MapReduce or Tez jobs. Lets analysts query HDFS data without writing Java.
• Pig: Dataflow scripting language that compiles to MapReduce.
• HBase: Column-family store (modeled on Google Bigtable) for random read/write access on top of HDFS.
• ZooKeeper: Distributed coordination service (leader election, distributed locks).
• Oozie: Workflow scheduler for chaining Hadoop jobs.

Hadoop's limitations drove the next generation. MapReduce writes intermediate results to disk between every stage (map output → shuffle → reduce input). For jobs with many stages (like machine learning algorithms that iterate over the data many times), this causes enormous I/O overhead. Each iteration touches the disk. Spark was built specifically to address this.

# Hive query (HiveQL) that compiles to a MapReduce job:
# Count unique users per country from a billion-row log table.
SELECT   country,
         COUNT(DISTINCT user_id) AS unique_users
FROM     access_logs
WHERE    dt = '2024-01-15'
GROUP BY country
ORDER BY unique_users DESC
LIMIT    20;

-- Hive translates this to:
-- MAP:    emit (country, user_id) for each row
-- SHUFFLE: group by country
-- REDUCE: COUNT(DISTINCT user_id) via HyperLogLog sketch
-- A second MR stage handles ORDER BY + LIMIT globally

Apache Spark (2009, by Matei Zaharia at UC Berkeley AMPLab; 2014 Apache top-level project) is the most important successor to Hadoop MapReduce. Its core abstraction is the Resilient Distributed Dataset (RDD).

An RDD is an immutable, partitioned, lazily evaluated collection of records that can be reconstructed from its lineage if a partition is lost. The three key ideas:

1. In-memory computation. Instead of materializing intermediate results to HDFS after every stage, Spark keeps data in memory across stages. For iterative algorithms (K-means, PageRank, gradient descent), each iteration reads from memory rather than disk — speedups of 10-100x are common.

2. Lazy evaluation. Spark operations are divided into transformations (map, filter, join, groupByKey) and actions (collect, count, saveAsTextFile). Transformations are lazy — they build a logical plan (a DAG of operations) but do not execute. Only an action triggers execution. This allows Spark's optimizer to see the full computation graph and perform optimizations like pipelining (fusing multiple map operations into a single pass) and stage planning (deciding where to insert shuffle boundaries).

3. Fault tolerance via lineage. Instead of checkpointing data (expensive), Spark remembers how each RDD was derived from its parent RDDs. If a partition is lost (because a machine died), Spark re-executes only the transformations needed to recompute that specific partition from the last checkpoint or original data source. This is elegant and cheap for narrow dependencies (one parent partition → one child partition), but expensive for wide dependencies (a shuffle where one child partition depends on all parent partitions).

Spark SQL and DataFrames (added in Spark 1.3) added a relational layer on top of RDDs with a query optimizer called Catalyst. This is now the dominant Spark API because the optimizer can push predicates down, reorder joins, and apply columnar storage tricks — often generating faster code than hand-written RDD operations.

from pyspark.sql import SparkSession

spark = SparkSession.builder.appName("WordCount").getOrCreate()
sc = spark.sparkContext

# RDD API: each line is a lazy transformation
lines = sc.textFile("hdfs:///data/shakespeare/*.txt")  # no execution yet
words = lines.flatMap(lambda line: line.lower().split())  # still lazy
pairs = words.map(lambda w: (w, 1))                       # still lazy
counts = pairs.reduceByKey(lambda a, b: a + b)            # still lazy

# THIS triggers execution — Spark builds the DAG and runs it:
top_20 = counts.sortBy(lambda x: -x[1]).take(20)
print(top_20)

# Iterative algorithm benefit: cache the dataset in memory
graph = sc.textFile("hdfs:///data/web_graph")\
           .map(parse_edge)\
           .cache()          # <-- tell Spark: keep this in RAM

# Now 50 PageRank iterations each read from RAM, not disk
for iteration in range(50):
    ranks = graph.join(ranks).map(compute_contribution).reduceByKey(sum)

When you trigger an action, Spark's DAG Scheduler analyzes the lineage graph of RDD transformations and divides it into stages. A stage boundary is created whenever there is a wide dependency — a transformation like groupByKey, reduceByKey, or join where the output partition depends on multiple input partitions (i.e., a shuffle is required).

Within a stage, operations are pipelined: multiple narrow transformations (filter, map, flatMap) on the same partition are fused into a single pass over the data without writing intermediate results to memory. This is the key performance advantage over Hadoop's one-disk-write-per-stage model.

The scheduler submits stages as sets of tasks (one task per partition). The Task Scheduler then assigns tasks to workers via the Cluster Manager (YARN, Kubernetes, or Spark's own standalone manager), preferring data-local execution.

Shuffle implementation in Spark. Spark's shuffle is similar to Hadoop's but improved. Map-side sort-and-spill writes sorted partition files to local disk. Reduce-side tasks fetch their partition files over the network. Later versions (Spark 3.x) introduced Adaptive Query Execution (AQE), which can dynamically merge small shuffle partitions, switch join strategies mid-query based on runtime statistics, and handle data skew automatically — problems that required manual tuning in earlier versions.

Understanding the DAG is crucial for debugging performance. The Spark UI's Stage DAG view lets you see exactly where shuffles occur, how many tasks each stage has, and which tasks are slow (stragglers). This is your primary tool for diagnosing slow Spark jobs.

# Visualizing Spark's stage boundaries
# Transformations marked [narrow] vs [wide] (shuffle boundary)

rdd1 = sc.textFile("logs/")               # Stage 1 starts here
rdd2 = rdd1.filter(is_error)              # [narrow] — pipelined
rdd3 = rdd2.map(parse_log_entry)          # [narrow] — pipelined
rdd4 = rdd3.map(lambda x: (x.host, 1))   # [narrow] — pipelined
# ---- SHUFFLE BOUNDARY (Stage 1 ends, Stage 2 begins) ----
rdd5 = rdd4.reduceByKey(lambda a,b: a+b) # [wide]  — triggers shuffle
rdd6 = rdd5.filter(lambda x: x[1] > 100) # [narrow] — pipelined
# ---- SHUFFLE BOUNDARY (Stage 2 ends, Stage 3 begins) ----
rdd7 = rdd6.sortByKey()                   # [wide]  — another shuffle

# Spark will create exactly 3 stages:
# Stage 1: textFile → filter → map → map → (shuffle write)
# Stage 2: (shuffle read) → reduceByKey → filter → (shuffle write)
# Stage 3: (shuffle read) → sortByKey → action

MapReduce and Spark (in its basic form) are batch processing systems: they operate on a bounded, finite dataset stored at rest. You accumulate data for a day (or an hour, or a week), then run a job over it. The result is available after the job completes — which might be minutes or hours after the data was generated.

Stream processing operates on unbounded, continuously-arriving data. Instead of waiting for a batch to accumulate, you process each event as it arrives (or in small micro-batches). The goal is low latency: ideally, results are updated within milliseconds to seconds of new data arriving.

The trade-off is not just latency vs. throughput:

• Batch has simple semantics. Results are exact because you have the complete dataset. Re-running a batch job gives the same answer. State management is trivial — you just read the dataset.
• Stream must handle late data (events that arrive after their logical time), out-of-order events, and the impossibility of ever seeing the "complete" picture. You must make decisions about watermarks (how long to wait for late events before closing a time window) and exactly-once semantics (hard to achieve without careful engineering of idempotent operations and transactional sinks).

Lambda architecture (Nathan Marz, 2011) is an influential pattern for having both: a batch layer for accurate historical results and a speed layer (stream processing) for fresh approximate results. The two are merged at query time. In practice, the complexity of maintaining two code paths for the same logic led to the Kappa architecture (Jay Kreps, 2014), which argues for a single stream processing system with enough reprocessing capability to handle the historical use case.

Apache Flink is the leading stream-first system, treating batch as a special case of streaming over a finite dataset. Spark Structured Streaming brings streaming to Spark using the same DataFrame API, treating a stream as an unbounded table.

# Spark Structured Streaming: word count over a live stream
# The API looks almost identical to batch Spark SQL
from pyspark.sql import SparkSession
from pyspark.sql.functions import explode, split

spark = SparkSession.builder.appName("StreamingWordCount").getOrCreate()

# Read from a Kafka topic (unbounded stream)
raw = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "log-events") \
    .load()

words = raw.selectExpr("CAST(value AS STRING)") \
    .select(explode(split("value", " ")).alias("word"))

counts = words.groupBy("word").count()

# Write running counts to console every trigger interval
query = counts.writeStream \
    .outputMode("complete") \
    .format("console") \
    .trigger(processingTime="10 seconds") \
    .start()

query.awaitTermination()

The file format you use for large-scale data matters enormously — sometimes as much as the compute engine itself. The dominant format in modern data systems is Apache Parquet, a columnar storage format.

Row-oriented storage (like CSV or a traditional row-store database) stores each row contiguously: all fields of record 1, then all fields of record 2, etc. This is optimal for workloads that need to read or write complete records (OLTP — order processing, user profile lookup).

Column-oriented storage stores each column contiguously: all values of column 1, then all values of column 2, etc. This is optimal for analytical queries that read a few columns from millions of rows — exactly the MapReduce/Spark workload. If your query is SELECT AVG(revenue) FROM sales WHERE year = 2023, columnar storage means you read only the revenue and year columns. With 100 columns, you read 2% of the data.

Columnar storage also enables dramatically better compression. Values in the same column tend to be similar (all revenues, all country codes), and compression algorithms exploit repeated patterns. Parquet uses a combination of dictionary encoding, run-length encoding, and delta encoding, often achieving 5-10x compression over row formats. Smaller files mean less network I/O in the shuffle and less storage cost.

Parquet also stores column statistics (min, max, null count per row group) in its footer. Spark and other engines exploit this for predicate pushdown: if your filter is year = 2023 and a row group's min/max year is 2020-2022, the entire row group can be skipped without reading it. This can reduce I/O by orders of magnitude on filtered scans.

The lesson is that storage format is a first-class performance concern in distributed systems, not an afterthought.

# Writing and reading Parquet with Spark (DataFrame API)
from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()

# Read a CSV (row format) and write as Parquet (columnar)
df = spark.read.csv("hdfs:///raw/sales.csv", header=True, inferSchema=True)
df.write \
  .partitionBy("year", "country")  # partition pruning on directory
  .parquet("hdfs:///processed/sales/")

# This query reads ONLY the 'revenue' and 'year' columns
# AND skips row groups where year != 2023 (predicate pushdown)
result = spark.read.parquet("hdfs:///processed/sales/") \
    .filter("year = 2023") \
    .agg({"revenue": "avg"})

result.explain()   # shows 'PushedFilters: [IsNotNull(year), EqualTo(year,2023)]'
                   # and 'PartitionFilters: [isnotnull(year), (year = 2023)]'

The number of partitions (the degree of parallelism) is the single most impactful tuning knob in a Spark or MapReduce job.

Too few partitions: you underutilize the cluster. If you have 200 executors but only 10 partitions, 190 executors sit idle.

Too many partitions: scheduling overhead dominates. Each task incurs overhead (JVM spinup, scheduler latency, shuffle file creation). Creating 100,000 tiny 1KB tasks is slower than 1,000 tasks of 100KB each.

The rule of thumb: aim for 2-4 tasks per CPU core, with each task processing roughly 100-200MB of data. For a 1TB shuffle, you would target around 2,000 partitions.

Data skew is the hardest performance problem in distributed processing. If 1% of your keys account for 50% of your data, the reducer (or executor) handling those keys will take 50x longer than others. Your entire job waits for that one slow task. Diagnosis: in the Spark UI, look for a stage where most tasks finish in 1 second but one task takes 10 minutes.

Solutions:

1. Salting. Append a random integer 0-N to hot keys before the shuffle, forcing them to split across N reducers. After aggregating, remove the salt and merge results.
2. Broadcast join. If one side of a join is small (< a few GB), broadcast it to all executors so the join happens locally without any shuffle at all.
3. Adaptive Query Execution (AQE, Spark 3.0+). Automatically detects skew and splits oversized partitions at runtime.
4. Custom partitioner. Pre-partition data by a business key (e.g., date) so joins on that key require no shuffle.

# Salting technique for skewed joins
import pyspark.sql.functions as F
from pyspark.sql import SparkSession

spark = SparkSession.builder.getOrCreate()

# Problem: user_id 42 appears in 40% of events
events = spark.table("events")
profiles = spark.table("user_profiles")

SALT_FACTOR = 20

# Replicate the small table (profiles) N times with all salt values
profiles_salted = profiles.withColumn(
    "salt", F.explode(F.array([F.lit(i) for i in range(SALT_FACTOR)]))
)
profiles_salted = profiles_salted.withColumn(
    "salted_key", F.concat(F.col("user_id"), F.lit("_"), F.col("salt"))
)

# Assign a random salt to each event row
events_salted = events.withColumn(
    "salt", (F.rand() * SALT_FACTOR).cast("int")
)
events_salted = events_salted.withColumn(
    "salted_key", F.concat(F.col("user_id"), F.lit("_"), F.col("salt"))
)

# Join on salted key — hot user_id is now spread across 20 partitions
result = events_salted.join(profiles_salted, on="salted_key").drop("salt", "salted_key")

MapReduce was a breakthrough, but its two-stage model (one map phase, one reduce phase, repeat) is inexpressive for complex pipelines. A system like PageRank requires dozens of MapReduce stages; a machine learning training loop requires hundreds. Each stage boundary means a shuffle and a disk write.

The generalization is the dataflow model: represent the computation as a directed acyclic graph (DAG) of operators, where data flows along edges and is processed by operators. Stages are fused wherever possible. This is exactly what Spark's DAG scheduler implements, and it is also the model behind Google's Dataflow (2015, formerly FlumeJava), Apache Beam, and Apache Flink.

Google Dataflow / Apache Beam introduced the Beam model for unified batch and stream processing. The key abstractions are PCollections (distributed datasets), PTransforms (operations), Windows (for grouping stream elements by time), and Triggers (for deciding when to emit windowed results). The same pipeline can run on the Dataflow managed service, on Spark, or on Flink, because Beam is an API layer above multiple runner engines.

Pregel (Google, 2010) is the graph processing model for problems like PageRank and connected components. It models computation as a series of supersteps, where in each superstep every vertex computes a new value based on messages from its neighbors. This vertex-centric model maps naturally to iterative graph algorithms that are awkward to express in MapReduce.

The key lesson of this evolution: the correct abstraction depends on the workload. Batch jobs on bounded data → Spark or Hive. Low-latency stream processing → Flink or Kafka Streams. Large-scale graph algorithms → Pregel or GraphX. SQL analytics on columnar data → BigQuery, Snowflake, or Spark SQL with Parquet. Real systems are built by choosing the right tool for the access pattern.

# Apache Beam unified batch/stream pipeline (Python SDK)
import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions

# This SAME pipeline runs on Dataflow (cloud), Spark, or Flink
# just by changing the runner option.
options = PipelineOptions(runner='DataflowRunner')  # or 'SparkRunner'

with beam.Pipeline(options=options) as p:
    (
        p
        | 'ReadLogs'   >> beam.io.ReadFromText('gs://bucket/logs/*.gz')
        | 'ParseEvent' >> beam.Map(parse_log_line)         # map
        | 'FilterErrors' >> beam.Filter(lambda e: e['level'] == 'ERROR')
        | 'ExtractHost'  >> beam.Map(lambda e: (e['host'], 1))
        | 'CountPerHost' >> beam.CombinePerKey(sum)        # reduce
        | 'FormatOutput' >> beam.Map(lambda kv: f"{kv[0]}: {kv[1]}")
        | 'WriteResults' >> beam.io.WriteToText('gs://bucket/output/')
    )
# Beam handles DAG optimization, shuffle, retries, and progress.