Transformations and Actions

"They kept asking me to map, filter, and join, and I kept nodding and writing it all down. I did not lift a finger until someone finally said count. Then I moved the whole cluster at once."

An RDD That Believes in Deadlines

By now you have seen that a Spark program describes a computation over a distributed collection without saying when any of it happens. Section 7.4 explained the mechanism behind that delay: Spark records your operations as a directed acyclic graph and waits before executing it. This section gives the two halves of the API their names and their costs. A transformation takes a distributed collection (an RDD or a DataFrame) and returns a new one, lazily, adding a node to the graph and computing nothing. An action asks for a concrete result, a count, a written file, a list of rows pulled back to the driver, and that request is what finally triggers execution of the whole recorded graph. Learn which operations fall on which side, and learn what each one costs, and you can read the performance of a Spark job off its source code before you ever run it.

1. Two Kinds of Operation, One Execution Model Beginner

The rule is mechanical and worth stating exactly. If an operation returns another distributed collection, it is a transformation and it is lazy: nothing runs, Spark just extends the lineage graph. If an operation returns something that is not a distributed collection, a number, a Python list, a written-out dataset, it is an action and it is eager: Spark compiles every transformation feeding into it and launches a distributed job. The return type is the tell. filter hands you back another RDD, so it waits; count hands you back an integer, so it runs.

Transformations themselves come in two flavors that matter enormously for cost. A narrow transformation needs only the records already on a partition to produce that partition's output: map applies a function to each record, filter drops records that fail a predicate, flatMap lets each record expand into zero or more outputs. No data crosses the network, so narrow transformations are nearly free relative to the alternative. A wide transformation needs records from many partitions to land together: groupByKey, reduceByKey, and join all gather every record sharing a key onto one partition, which means moving data across machines. That movement is the shuffle, the same expensive all-to-all redistribution you met as the heart of MapReduce in Chapter 6, and it is the dominant cost in most Spark jobs. Table 7.5.1 lays out the common operations on both axes.

2. The Aggregation Choice: reduceByKey Versus groupByKey Intermediate

Two wide transformations can compute the same per-key aggregate, and choosing between them is the highest-leverage decision in this section. Suppose you have billions of (key, value) pairs and you want one summary per key, a sum, a mean, a count. The naive path is groupByKey: it shuffles every single record so that all values for a key arrive together on one partition, and only then do you reduce them. The efficient path is reduceByKey: it applies your combining function twice, once locally on each partition before the shuffle and once globally after, so each partition ships only one partial result per key instead of every raw record.

This map-side combine is exactly the combiner optimization from Section 6.4, now baked into a single API call. The mathematics that licenses it is that your reduction must be associative and commutative, so that partial reductions can be combined without changing the answer. For a commutative-associative operator $\oplus$ over the values $v_1, \dots, v_m$ of one key spread across partitions $\mathcal{P}_1, \dots, \mathcal{P}_p$,

the inner reductions run locally before any network traffic, and only the $p$ partial results per key cross the wire. The reduction in bytes shuffled is dramatic whenever the number of distinct keys is far smaller than the number of records, which is the common case for aggregation. Figure 7.5.1 contrasts the two dataflows.

The demo below makes the saving concrete with no cluster required. It builds a million (key, value) pairs spread across four partitions with only twenty distinct keys, then accounts for the bytes each strategy would push through the shuffle: groupByKey sends every record, while reduceByKey first collapses each partition to one (sum, count) partial per local key and sends only those. It also checks that both paths produce the same per-key mean, so the saving is free of any accuracy cost.

import random
random.seed(7)

# A dataset of (key, value) pairs spread across P map partitions.
# Few distinct keys, many records per key: the classic aggregation shape.
P, N_PER_PART = 4, 250_000
KEYS = [f"user_{i:02d}" for i in range(20)]          # 20 distinct keys
partitions = [[(random.choice(KEYS), random.random())
               for _ in range(N_PER_PART)] for _ in range(P)]

VAL_BYTES = 8                                          # an 8-byte float value
kbytes = lambda k: len(k.encode("utf-8"))             # key size in bytes

# groupByKey: every raw record is shuffled to the reducer that owns its key.
group_bytes = sum(kbytes(k) + VAL_BYTES
                  for part in partitions for k, _ in part)

# reduceByKey: each partition first reduces to ONE (sum, count) partial per
# local key, then ships only those partials. This is the map-side combine.
PARTIAL_BYTES = 8 + 8                                  # (sum: float, count: int64)
reduce_bytes = 0
for part in partitions:
    local = {}
    for k, v in part:                                 # local reduce, no network
        s, c = local.get(k, (0.0, 0)); local[k] = (s + v, c + 1)
    reduce_bytes += sum(kbytes(k) + PARTIAL_BYTES for k in local)

records total          : 1,000,000
distinct keys          : 20
map partitions         : 4
groupByKey  shuffled   :    15.00 MB  (every record)
reduceByKey shuffled   :   0.0018 MB  (one partial per key per partition)
shuffle reduction      :     8152x less data moved
max abs diff in means  : 6.27e-15  (identical result)

The lesson generalizes past these two calls. Whenever an aggregation can be expressed as a commutative-associative reduction, prefer the variant that combines on the map side: reduceByKey over groupByKey, aggregateByKey with a sensible local merge, or in the DataFrame API a groupBy().agg() over built-in functions, which Spark's optimizer compiles into a partial-aggregate-then-final-aggregate plan automatically. The eight-thousand-fold gap in Output 7.5.1 is not an exotic edge case; it is the everyday difference between a job that fits in cluster memory and one that spills, stalls, or dies.

3. Actions and Their Costs Beginner

Actions are where the job actually happens, and they differ wildly in what they cost. The cheapest are the bounded ones. take(n) and first ask for only a handful of rows, and Spark is smart enough to evaluate just enough partitions to satisfy the request rather than scanning everything, so they are the right tools for peeking at data during development. count is more expensive because it must visit every partition, but it returns a single integer, so it is safe for the driver regardless of how large the dataset is.

The action to respect is collect. It pulls every row of the distributed collection back to the single driver process, which means the entire dataset must fit in the driver's memory at once. On a dataset that lives across a hundred machines precisely because it does not fit on one, collect asks the driver to do the impossible, and the result is an out-of-memory crash that takes down the whole application. The discipline is simple: never call collect on data you have not already shrunk to driver scale, with a filter, an aggregation, or a take. When you want to persist a large result, use a writing action such as write or saveAsTextFile, which keeps the output distributed across the cluster and never funnels it through the driver.

4. How a Chain Becomes One Job Intermediate

The payoff of laziness is that a long chain of transformations followed by a single action compiles into one optimized job rather than a sequence of materialized steps. Because Spark holds the whole transformation graph before any action fires, it can fuse adjacent narrow operations into a single pass over the data, push filters earlier so less data flows downstream, and choose how to execute each wide operation, all before the first byte moves. The boundaries between the resulting stages fall exactly at the shuffles: a run of narrow transformations becomes one stage that needs no network, and each wide transformation forces a stage boundary because data must be redistributed before the next run of work can begin. Section 7.6 shows how to keep an intermediate result in memory so repeated actions do not recompute the chain, and Section 7.4 covered the DAG machinery that makes this fusion possible.

# PySpark, shown illustratively. logs is an RDD of raw request-log lines.
result = (logs
          .filter(lambda line: "ERROR" in line)        # narrow: no shuffle
          .map(lambda line: (parse_service(line), 1))  # narrow: no shuffle
          .reduceByKey(lambda a, b: a + b)             # wide: ONE shuffle here
          .filter(lambda kv: kv[1] > 100))             # narrow: no shuffle

top = result.take(10)   # the FIRST action: only now does any of the above run

Read Code 7.5.2 the way Spark does. The first three lines and the last filter add nodes to a graph and compute nothing. The take(10) is the action that compiles the graph into a two-stage job: stage one fuses the filter and map and performs the map-side combine of the reduceByKey, then shuffles the small partials; stage two merges the partials, applies the final filter, and returns ten rows. The single reduceByKey rather than a groupByKey is what keeps that shuffle tiny, by exactly the factor Output 7.5.1 measured.

from pyspark.sql import functions as F

errors_per_service = (df
    .filter(F.col("level") == "ERROR")
    .groupBy("service")
    .agg(F.count("*").alias("n"))   # Catalyst plans partial + final aggregate
    .filter(F.col("n") > 100))

errors_per_service.show()           # the action that triggers the job

5. Reading the Cost Off the Code Intermediate

The vocabulary of this section turns into a habit you can apply to any Spark program before running it. Scan for the actions first: they are the only lines that move the cluster, and counting them tells you how many jobs you are launching (re-running the same chain for several actions without caching recomputes it each time, a waste Section 7.6 fixes). Then, between actions, find the wide transformations: each is a shuffle and therefore a stage boundary and a network cost, while the narrow ones in between are nearly free. Finally, audit every aggregation for the reduceByKey-versus-groupByKey choice, and every collect for whether the data has really been shrunk to driver scale. Those three passes, actions, shuffles, and the combine-or-not decision, predict the performance of most Spark jobs without a profiler.

This way of reading code also explains why the performance models of Chapter 3 apply so directly to Spark. The narrow transformations are the parallel computation that scales with more machines; the shuffles are the communication term that does not. A job dominated by narrow work scales out beautifully; a job dominated by a giant groupByKey shuffle is communication-bound, and adding machines helps little until you shrink the shuffle. The cost you read off the code is the cost the model predicts.

You now have the two halves of the Spark API and their costs: transformations record intent lazily, actions force execution eagerly, and the choice to combine on the map side before a shuffle is the difference between a job that scales and one that chokes. The next question is how to control where data lives across partitions and how to avoid recomputing a chain you need more than once. That is the work of partitioning and caching, which Section 7.6 takes up next.