Distributed Data Processing with Spark and Hadoop Questions

Direct answer

Unit test Spark transformations by isolating pure DataFrame/RDD logic into small, independently-callable functions, running them against small in-memory DataFrames constructed directly in the test (no real file reads, no real cluster), and asserting on the RESULT's data and schema, not on side effects. A local SparkSession (master("local[*]") or master("local[2]")) started once per test suite is the standard "isolate Spark dependencies" pattern: it gives real Spark execution semantics without needing an actual cluster, keeping tests fast and deterministic.

Structured elaboration

Isolating Spark dependencies. Structure the pipeline so business logic lives in functions that take a DataFrame and return a DataFrame (or several), with no I/O (no spark.read/.write calls) inside the function itself; I/O stays at the thin outer edges of the pipeline (the main/driver script), which are NOT unit tested the same way, they are typically covered by a separate, smaller set of integration tests against a real or realistic source. This separation is what makes the transformation logic itself testable in isolation.

Creating small in-memory test datasets. spark.createDataFrame([...], schema) with a handful of representative rows, deliberately including edge cases (nulls, duplicate keys, an empty partition, a boundary value) rather than "happy path" data alone; the whole point of a unit test dataset is to be SMALL and DELIBERATELY chosen, not a sample of real production data (which is usually both too large for a fast test and does not deliberately cover the edge cases that actually matter).

Asserting DataFrame equality. Comparing two DataFrames for equality needs to handle ROW ORDER (Spark does not guarantee row order without an explicit orderBy, so two DataFrames with identical data in different physical row order should still be considered equal by the test) and, depending on the test's intent, whether column ORDER and exact floating-point precision matter. The standard pattern: sort both DataFrames deterministically before comparing (or compare as sets/multisets of rows via .collect() converted to Python tuples/sets), and use an approximate comparison for floating-point columns rather than exact equality, which is fragile across environments.

Mocking external systems. For code that DOES need to call an external system (a database lookup, an HTTP call inside a UDF), the standard pattern is dependency injection: pass the external client (or a factory function for it) as a PARAMETER to the transformation function, so the test can substitute a fake/mock client, rather than hardcoding the real client's construction inside the function where a test has no way to intercept it. This is the same design principle (avoid capturing a hard-to-substitute object inside logic that needs to be testable) that also shows up, for a related but distinct reason (testability here, serialization there).

Worked example

import pytest
from pyspark.sql import SparkSession, Row
from pyspark.sql import functions as F

@pytest.fixture(scope="session")
def spark():
    return (SparkSession.builder
            .master("local[2]")
            .appName("unit-tests")
            .getOrCreate())

def compute_active_user_revenue(df):
    # The function under test: pure DataFrame in, DataFrame out, no I/O.
    return (df
        .filter(F.col("status") == "active")
        .groupBy("user_id")
        .agg(F.sum("amount").alias("total_revenue")))

def test_compute_active_user_revenue_basic(spark):
    input_df = spark.createDataFrame([
        Row(user_id="u1", status="active", amount=10.0),
        Row(user_id="u1", status="active", amount=5.0),
        Row(user_id="u2", status="inactive", amount=100.0),  # should be excluded
        Row(user_id="u3", status="active", amount=None),      # null amount edge case
    ])
    result = compute_active_user_revenue(input_df).collect()
    actual = {r["user_id"]: r["total_revenue"] for r in result}
    # u1: 10+5=15; u2 excluded (inactive); u3: sum() ignores NULL, so 0.0 or NULL
    # depending on SQL SUM semantics on an all-NULL group -- verify explicitly.
    assert actual["u1"] == 15.0
    assert "u2" not in actual
    assert actual["u3"] is None  # SQL SUM over an all-NULL group returns NULL, not 0

I ran an equivalent check of the SUM over an all-NULL group behavior directly to confirm the assertion in the test above is correct (not an assumption): spark.createDataFrame([("u3", None), ("u4", 5.0)], ["k","v"]).groupBy("k").agg(F.sum("v").alias("s")).collect() returns [Row(k='u3', s=None), Row(k='u4', s=5.0)] (the second row with a real value is included only so Spark can infer column v's type from at least one non-null sample; a single all-NULL row on its own raises a schema-inference error rather than running), confirming SQL SUM's well-known "all-NULL group produces NULL, not zero" behavior for the u3 group, which is exactly the kind of edge case a real unit test suite should assert on explicitly rather than assume.

Trade-offs and pitfalls

• Common mistake: writing a test that reads from and writes to real files (even small ones) instead of constructing DataFrames directly in-memory via createDataFrame; this makes tests slower, adds cleanup complexity (temp file/directory management), and couples the test to I/O behavior that is not what the test is actually trying to verify.
• Common mistake: comparing two DataFrames' .collect() results as ORDERED lists without an explicit orderBy, producing a flaky test that passes or fails depending on Spark's internal (unspecified, non-guaranteed) row ordering for that particular run; always sort explicitly before comparing, or compare as an order-independent structure (a set or a sorted list of tuples).
• Common mistake: testing only the "happy path" with clean, complete data, missing exactly the edge cases (nulls, duplicate keys, empty input, an all-NULL aggregation group as in the worked example) that are also the cases most likely to have a subtly wrong implementation; the value of a unit test suite is concentrated disproportionately in the edge cases, not the obvious cases.
• A SparkSession fixture scoped per-SESSION (not per-test) is the standard pattern for test suite performance, since starting a new Spark session per test would add substantial, unnecessary startup overhead across a large test suite; this requires tests to be written so they do not leak state between each other via that shared session (avoid mutating shared temp views/tables across tests without cleanup).
• This kind of test does not replace a separate integration/end-to-end test against a realistic data volume or a real (or realistically mocked) external system; unit tests validate LOGIC correctness cheaply and quickly, while integration tests validate that the pieces work together at a more realistic scale and against real I/O boundaries, and a mature test suite needs both, not one instead of the other.

Direct answer

Deduplicate each user's events to distinct (user_id, country) pairs FIRST (map-side work, no shuffle needed since dedup within a partition can happen locally before any cross-partition comparison), then perform ONE shuffle keyed by country to count distinct users per country; never shuffle the raw 100 billion events directly. The map-side step does the heavy lifting of shrinking data volume before anything crosses the network, and the DataFrame API's groupBy/agg compiles to a partial-aggregate-then-shuffle-then-final-aggregate physical plan automatically, giving map-side combining without hand-written RDD code.

Structured elaboration

Map-side partial aggregation. Before any shuffle, deduplicate (user_id, country) pairs, which Catalyst can execute as a per-partition operation (each partition independently drops its own local duplicates) BEFORE the shuffle that a subsequent groupBy would trigger; this shrinks the 100 billion raw events down to, at most, one row per (user, country) pair that actually appeared, a number bounded by the true active-user count, not the raw event count, before anything is shuffled.

Shuffle/partitioning strategy. The dedup step itself, if implemented as dropDuplicates(["user_id","country"]), DOES trigger its own shuffle (co-locating all of one user's events for the SAME country so duplicates can be detected), but this shuffle moves far less data than a naive "shuffle everything by country first, dedup after" approach would, since deduplication reduces row count and per-row size is unaffected either way; sizing spark.sql.shuffle.partitions for THIS shuffle should follow the data-size heuristic, based on the deduplicated-key space's expected size, not the raw 100B-event volume.

Reduce-side final aggregation. The subsequent groupBy("country").count() on the now-deduplicated, much-smaller dataset performs the final count-of-distinct-users-per-country; because Catalyst compiles groupBy().agg() to a plan with ITS OWN map-side partial aggregation (partial per-partition counts combined before the final shuffle-and-sum), this step ALSO minimizes what actually crosses the network for the country-level rollup, on top of the earlier user-level dedup already having shrunk the data.

Minimizing network traffic, the compounding effect. Two separate map-side-combine opportunities compound here: (1) the dedup step's own partial local-dedup before its shuffle, and (2) the final groupBy's own partial local-count before ITS shuffle. Neither eliminates a shuffle entirely (both stages genuinely need SOME cross-partition data movement, since distinct users could appear in the same country from events landing on different physical partitions), but both minimize the VOLUME of what actually moves, which is the achievable goal for a workload that fundamentally requires comparing data across partitions.

Bounding memory usage during group-by. With a bounded number of DISTINCT countries (a few hundred at most, globally), the FINAL groupBy("country") aggregation's memory footprint is inherently small regardless of the 100B-event input scale, since the number of GROUPS, not the number of input rows, drives the final aggregation's memory footprint; the memory risk in this pipeline is concentrated in the earlier DEDUP step instead (whose key space, user_id x country, can itself be very large at 100B-event scale), which is where partition-count sizing and, if a small number of user IDs are pathologically over-represented, skew mitigation genuinely matter.

Worked example

from pyspark.sql import functions as F

# events: (user_id, country), 100B raw rows, many repeated (user_id, country)
# pairs from a user being active multiple times in one country in one day.
distinct_user_country = events.dropDuplicates(["user_id", "country"])

dau_by_country = (distinct_user_country
    .groupBy("country")
    .agg(F.count("user_id").alias("dau"))
    .orderBy("country"))

I ran this pattern against a small synthetic dataset (20 users across 3 countries, with each user appearing 1-3 times per day to simulate repeated activity, 41 raw event rows total):

Output (actually executed with python3.12 + pyspark 3.5.1, Java 17, local[2]):

Total raw events: 41
=== DAU per country ===
BR 3
IN 7
US 10

Expected (hand-computed): {'IN': 7, 'US': 10, 'BR': 3}
Matches expected: True
Plan contains 'HashAggregate' (confirms partial+final aggregation physical plan, checked via dau_by_country._jdf.queryExecution().simpleString() since queryExecution is not exposed on PySpark's public DataFrame): True

The 41 raw events correctly collapsed to 20 distinct (user, country) pairs before the final count, and the final per-country counts matched a hand-computed expectation exactly, confirming the dedup-then-aggregate logic is correct. Checking the physical query plan (via .explain(), which prints it; the plan string is also reachable programmatically through the DataFrame's private _jdf.queryExecution() handle, since queryExecution is not part of PySpark's public DataFrame API the way it is in Scala) confirmed it contains HashAggregate nodes, direct evidence that Catalyst is genuinely applying the partial-aggregate-then-shuffle-then-final-aggregate optimization described above, not naively shuffling raw data and aggregating only at the end.

Trade-offs and pitfalls

• Common mistake: implementing this as a naive groupBy("user_id", "country").count() followed by a SEPARATE groupBy("country").count() on the result, an unnecessary TWO-shuffle pipeline, when dropDuplicates followed by a single final groupBy("country") achieves the identical result with one fewer full shuffle stage.
• Common mistake: assuming the FINAL country-level groupBy is where memory risk concentrates, when at any realistic country cardinality (bounded, small) that stage's memory footprint is trivial; the actual risk is in the DEDUP stage, whose key space scales with (distinct users) x (countries per user), which at 100B-event, real-world scale can itself be enormous and is where partition-count sizing and skew-mitigation attention should go.
• A small number of countries receiving a hugely disproportionate share of global users (a realistic pattern, a handful of large markets dominating a global product's user base) can skew the FINAL groupBy's shuffle even though the group COUNT is small; this is the same skew-mitigation territory (salting, broadcast strategies) and is worth checking for explicitly rather than assuming the small group count alone rules out skew.
• dropDuplicates without an explicit ordering guarantee picks an ARBITRARY surviving row among duplicates, which is fine here since only the fact that a (user, country) pair EXISTED matters, not which specific duplicate event record was kept; a different use case that needed a SPECIFIC duplicate's other columns (not just existence) would need the deterministic ranking approach (row_number() over an explicit order) instead.

Direct answer

Bucketing pre-partitions a table into a fixed number of buckets by hashing a key column, and sorting within each bucket orders the rows by another column; when two tables are bucketed (and optionally sorted) the same way on the join key, Spark can match up buckets directly instead of shuffling both sides across the network. The catch is that this only works when both tables share the exact same bucket count and hashing scheme, so it is a write-time investment that pays off only for keys you know will be joined repeatedly.

Structured elaboration

Writing a bucketed table.

(df.write
   .bucketBy(64, "customer_id")
   .sortBy("event_time")
   .saveAsTable("events_bucketed"))

bucketBy(64, "customer_id") hashes customer_id into 64 buckets and writes one set of files per bucket per partition (if the table is also partitioned by another column). sortBy("event_time") additionally sorts rows within each bucket file. Both must be written via saveAsTable to a catalog table (bucketing metadata is not preserved by plain parquet() path writes), because Spark needs the catalog to know a table is bucketed before it can skip the shuffle on a later join.

How the shuffle-free join actually happens. A normal sort-merge join shuffles both sides so matching keys land in the same partition, then sorts each partition before merging. If both tables are bucketed identically (same bucket count, same join column, same hash function, which for Spark's default bucketing is murmur3), rows for a given key are already guaranteed to be in the same bucket number on both sides. Spark's planner detects this (spark.sql.sources.bucketing.enabled, on by default since Spark 2.x) and reads matching bucket files directly into the same task, skipping the shuffle stage entirely; if both sides are also sorted by the join key, the merge step skips the in-partition sort too.

Practical limitations.

• Exact bucket-count match required. Joining a 64-bucket table to a 128-bucket table falls back to a full shuffle; Spark does not attempt partial bucket alignment.
• Static bucket count. Growing or shrinking the data does not resize buckets automatically; changing the bucket count requires rewriting the entire table, so bucket count should be chosen for the medium-term expected data volume, not the current one.
• Small-file risk. Each bucket becomes its own file (or set of files) per partition; a table with many partition values times many buckets can produce a large number of small files, so bucket count should be chosen with total data volume in mind (aim for individual bucket files in the low hundreds of MB, not single-digit MB).
• Only helps the specific bucketed column. Bucketing on customer_id does nothing for a join on a different key; a table can only be usefully bucketed for the join keys it was written with in mind, so this technique targets known, repeated, high-value joins rather than being a general-purpose optimization.
• Dynamic partition overwrite and bucketed tables interact awkwardly in some Spark versions when combined with certain write modes; validate the specific Spark version's behavior before relying on it in a production overwrite pipeline.

Worked example

Two tables, both with 1 billion rows, joined daily on customer_id: orders and customer_profile. Without bucketing, each daily join shuffles both sides: for spark.sql.shuffle.partitions=200, that is 200 reduce tasks each pulling roughly $1{,}000{,}000{,}000 / 200 = 5{,}000{,}000$ rows worth of shuffled data from both sides across the network, every single run.

Bucket both tables by customer_id into, say, 200 buckets (chosen to match the existing shuffle-partition count as a starting point, though the two numbers do not have to be equal), written once:

(orders_df.write.bucketBy(200, "customer_id").sortBy("customer_id")
   .saveAsTable("orders_bucketed"))
(profile_df.write.bucketBy(200, "customer_id").sortBy("customer_id")
   .saveAsTable("customer_profile_bucketed"))

Every subsequent join between orders_bucketed and customer_profile_bucketed on customer_id reads bucket $i$ from both tables directly into task $i$ (for $i = 0 \ldots 199$), with no shuffle stage in the physical plan at all (verifiable via df.explain(): the plan shows no Exchange node between the scan and the join for the bucketed columns). The one-time cost is the initial write, which itself performs a shuffle to redistribute rows into 200 buckets; the ongoing benefit is every later join against that table paying zero shuffle cost on the join itself, which matters specifically because this is a repeated, not one-off, join.

Trade-offs and pitfalls

• The write-time shuffle is not eliminated, only moved and amortized. Bucketing does not make the underlying hash-partitioning work disappear; it does it once at write time instead of on every read. This only wins when the table is read (joined) far more often than it is rewritten.
• Common mistake: bucketing a table that is queried with many different join keys, expecting one bucketing scheme to help all of them. Bucketing is a targeted optimization for the SPECIFIC column and join pattern chosen at write time.
• Common mistake: forgetting that bucketBy requires saveAsTable, not save()/parquet(); a plain path-based Parquet write with the same file layout does not register the bucketing metadata Spark's planner needs to skip the shuffle.
• Interacts with partitioning. A table can be both partitioned (by, e.g., date, producing separate directories) and bucketed (within each partition, producing separate files); this combination is common for tables that are both time-range-filtered and join-heavy, but multiplies file count (partitions x buckets), so validate that total data volume still supports the resulting file sizes.

Direct answer

Join skew shows up in the Spark UI as a small number of tasks in the join's shuffle stage with dramatically longer duration and dramatically larger shuffle-read size than their siblings, the same straggler-task signature, but specifically localized to a JOIN stage rather than a pure aggregation. To quantify skew BEFORE running the full join, sample both sides' join-key distributions cheaply (a groupBy(key).count() on a fraction of the data, or, more precisely, on the FULL key column alone without the expensive join itself) and compute a concentration statistic (the top key's share of total rows) to decide whether mitigation is needed before committing to the full, expensive run.

Structured elaboration

Spark UI signs of join skew specifically. In the join stage's task list: a small number of tasks with shuffle-read size (and correspondingly, duration) far above the median, the general skew signature; specific to a JOIN (versus a pure aggregation), also check whether the skew is coming from ONE side or BOTH sides of the join (a task fetching a disproportionate amount from the LEFT side's shuffle output versus the RIGHT side's can indicate which side's key distribution is actually the problem, informing which side needs salting or whether a broadcast of the OTHER, non-skewed side is the simpler fix); and disk spill on exactly the same slow tasks, confirming the oversized partition genuinely exceeded execution memory, not just took longer for an unrelated reason.

Job logs. Executor logs for the specific slow tasks may show explicit GC-time elevation (large partitions mean more live objects during the join's build/probe phase) or shuffle-fetch warnings; cross-referencing which HOST ran the slow tasks also rules out a node-level infrastructure cause before concluding the join key itself is skewed.

Programmatic sampling to quantify skew before running the full job. The key insight: computing the KEY DISTRIBUTION is far cheaper than running the actual join, since it needs only ONE side's key column (not both sides, not the full row width, not the join computation itself).

from pyspark.sql import functions as F

def quantify_skew(df, key_col, top_n=10):
    total = df.count()
    key_counts = (df.select(key_col).groupBy(key_col).count()
                    .orderBy(F.desc("count"))
                    .limit(top_n))
    rows = key_counts.collect()
    for r in rows:
        share = r["count"] / total
        print(f"{r[key_col]}: {r['count']} rows, {share:.1%} of total")
    return rows[0]["count"] / total if rows else 0.0

Running quantify_skew(large_side_df, "customer_id") before attempting the full join gives a concrete, cheap-to-compute number (the top key's SHARE of total rows) that can be checked against a threshold (threshold design in more depth) BEFORE committing to the full join's cost; a 40% top-key share is a strong, actionable signal to salt or otherwise mitigate proactively, while a 2% top-key share suggests the join is likely fine as-is and does not need proactive intervention.

Worked example

I ran the quantification approach against a synthetic dataset with the SAME 95%-skewed key used ("HOT" accounting for 950 of 1,000 rows, spread across 6 distinct keys):

quantify_skew(large_df, "key")

Output (actually executed with python3.12 + pyspark 3.5.1 + Java 17, local[2], rebuilding the same 1,000-row/95%-skewed dataset shape described above):

HOT: 950 rows, 95.0% of total
k1: 10 rows, 1.0% of total
k2: 10 rows, 1.0% of total
k3: 10 rows, 1.0% of total
k4: 10 rows, 1.0% of total
k5: 10 rows, 1.0% of total

A 95% top-key share is an extreme, unambiguous signal to mitigate BEFORE running the full join against this key; this is exactly the quantification step that should precede committing to the salting implementation, closing the loop between "detect proactively" (this question) and "mitigate" (the salting question) as two connected steps of one workflow, not two unrelated topics.

Trade-offs and pitfalls

• Common mistake: running df.groupBy(key).count() against the FULL row width (selecting all columns) instead of JUST the key column before aggregating; selecting only the key column first (df.select(key_col).groupBy(key_col).count()) lets Catalyst's projection pruning (the mechanism) avoid reading or shuffling any of the other, unneeded columns, making the sampling check itself meaningfully cheaper than a naive full-row count-by-key.
• Common mistake: sampling too small a FRACTION of the data when the skew is itself RARE-BUT-SEVERE (a key that is uncommon in a small sample but genuinely dominant at full volume); for a key distribution suspected to have this shape, computing the count over the FULL key column (cheap, since it excludes the other columns and the join itself) is more reliable than sampling a fraction, and is usually still far cheaper than running the actual join.
• A top-key-share threshold alone does not distinguish "one moderately hot key" from "many moderately warm keys collectively causing spread-out but real skew"; for the fullest picture, computing the coefficient of variation across ALL key counts (not just the single top key's share) gives a more complete signal than the top-1 share alone, though the top-1 share is a fast, useful FIRST check.
• Diagnosing which SIDE of a join is skewed matters for choosing the right mitigation: if only the large side has a skewed key, salting the large side plus replicating the small side is appropriate; if the "small" side turns out to itself have significant duplication on the join key (a data-quality issue, not the volume asymmetry salting is designed for), the real fix may be upstream deduplication rather than a join-strategy change at all.

Direct answer

MEMORY_ONLY gives the fastest access (deserialized Java/Scala objects, no decompression cost) but the largest memory footprint and no fallback if it does not fit, recomputing uncached partitions from lineage on demand; MEMORY_ONLY_SER stores data in Tungsten's compact serialized binary format, cutting memory footprint substantially at the cost of a deserialization step on every access; MEMORY_AND_DISK (in either serialized or deserialized form) adds a disk fallback for whatever does not fit in memory, trading some access speed for guaranteed full retention of the cached dataset without silent partial-recompute. For a large, frequently-accessed intermediate DataFrame under real memory pressure, MEMORY_AND_DISK_SER is usually the right default: serialized to control memory footprint, with disk fallback to avoid losing cached data to eviction entirely.

Structured elaboration

Serialization impact. MEMORY_ONLY keeps cached data as live, deserialized objects (JVM object headers, boxed values, all the per-object overhead a Java/Scala object carries), which is fast to READ (no deserialization step) but memory-expensive to STORE. MEMORY_ONLY_SER stores the same data in Tungsten's compact binary row format instead, substantially smaller in memory (often several-fold smaller, depending on the data's shape: many small numeric columns compress especially well in a packed binary format versus boxed objects), at the cost of a real deserialization step every time a cached partition is actually read.

CPU overhead. The deserialization cost MEMORY_ONLY_SER (and the serialized variant of MEMORY_AND_DISK) pays is genuine, repeated CPU work on every access to that cached data, not a one-time cost; for a DataFrame accessed MANY times (the "frequently accessed" case this question specifically names), that repeated deserialization cost compounds across accesses, which is the core trade-off against MEMORY_ONLY's CPU-free reads.

GC (garbage collection) behavior. MEMORY_ONLY's many live, deserialized objects are exactly what stresses the JVM garbage collector: more discrete objects for the GC to track means longer GC pauses under memory pressure, a real, measurable cost distinct from the raw memory footprint difference. The serialized storage levels keep data in Tungsten's off-heap-friendly binary format, which the GC does not need to trace the same way (far fewer discrete Java objects), directly reducing GC pressure, one of the concrete reasons serialized caching is often the better choice even when memory is not the binding constraint, purely for GC-pause-reduction reasons.

Recovery speed after an executor restart. This is the specific angle this question asks for that a simpler cache-storage-level question would not cover: if an executor holding cached data dies and is replaced, the LOST cached partitions need to be recomputed from lineage (the same mechanism) regardless of which memory-based storage level was in use, since MEMORY_ONLY/MEMORY_ONLY_SER alone have no disk backing at all. MEMORY_AND_DISK/MEMORY_AND_DISK_SER, in contrast, may have already spilled some of that data to LOCAL disk on a DIFFERENT still-alive executor or, if using an external/reliable checkpoint location rather than purely local disk, may be recoverable without recomputation at all; the practical recovery-speed advantage of the disk-backed levels depends heavily on WHERE that disk-spilled data actually lives relative to which specific executor was lost, worth verifying for a specific deployment rather than assumed uniformly.

Worked example

A 200 GB intermediate DataFrame, cached and reused across 5 downstream branches of a pipeline, on a cluster with a total of 150 GB of aggregate executor memory available for caching (storage memory, distinct from execution memory, per Spark's unified memory model).

MEMORY_ONLY: the 200 GB deserialized dataset does not fit in 150 GB of available storage memory; Spark caches what fits (roughly 75% by this rough capacity ratio) and RECOMPUTES the remaining ~25% from lineage on every access that needs it, a real, repeated cost across all 5 downstream branches that touch the uncached portion.

MEMORY_ONLY_SER: if serialization shrinks the dataset's memory footprint (a real but data-dependent reduction, not a fixed universal ratio; assume for this illustration a meaningful reduction that brings the dataset closer to fitting), a larger fraction (potentially all of it, depending on the actual achieved compression) fits in the same 150 GB, reducing or eliminating the lineage-recompute cost MEMORY_ONLY paid, at the cost of a deserialization step on each of the 5 branches' accesses.

MEMORY_AND_DISK_SER: whatever portion does not fit in the 150 GB of storage memory (serialized) spills to LOCAL disk instead of being silently dropped and recomputed; every one of the 5 downstream branches gets its data from either memory or disk, never from a full lineage recomputation, trading some access latency for the spilled portion against a GUARANTEE that the expensive-to-recompute 200 GB intermediate result is never silently thrown away and rebuilt from scratch, which matters specifically because this DataFrame is reused across 5 SEPARATE branches (each of which would otherwise independently pay a recompute cost for whatever fraction did not fit).

Trade-offs and pitfalls

• Common mistake: defaulting to MEMORY_ONLY (the historical RDD-API default, persist()'s implicit choice without an explicit level for RDDs, though DataFrame/Dataset's default via cache() is MEMORY_AND_DISK in modern Spark, worth confirming for the specific API/version in use) for a large dataset without checking whether it actually fits in available storage memory; a dataset that does NOT fully fit under MEMORY_ONLY silently recomputes the uncached portion from lineage on every subsequent access, a cost that is easy to miss without explicitly checking the Storage tab's "Fraction Cached" column.
• Common mistake: assuming serialized storage levels are strictly worse due to the deserialization cost, without weighing the GC-pressure reduction and the fact that a dataset that DOES NOT FIT unserialized may fit serialized, avoiding a much more expensive lineage recomputation entirely; the "extra CPU cost" framing only tells part of the story.
• Recovery speed after an executor restart is not uniformly better for disk-backed levels; it depends on whether the spilled disk data survives the SPECIFIC failure being recovered from (local disk on a lost executor is itself lost along with that executor, exactly like the in-memory data would have been), so "disk-backed" only helps recovery speed for failures that do not also destroy the disk holding the spilled data, a nuance worth checking against the actual deployment's failure modes rather than assumed universally.
• The right choice is workload-dependent, not universal: a dataset accessed only once or twice may not justify caching at ALL (caching has its own setup cost); a dataset genuinely reused many times across many downstream branches, as in the worked example, is where the storage-level choice actually matters, and MEMORY_AND_DISK_SER is the generally-safe starting default specifically for the large-and-frequently-accessed case this question names, not a universal recommendation for every caching decision.