Distributed Data Processing with Spark and Hadoop Questions

Direct answer

Implement getPartition to hash only the customer-ID PREFIX of each key (not the full composite key), so every key sharing that prefix, regardless of what follows it, routes to the same partition index; this colocates all of one customer's records in a single partition, letting a downstream join or windowed computation on that customer skip the shuffle it would otherwise need. Spark's Python API does not expose a Partitioner CLASS the way Scala does, but achieves the identical mechanism via rdd.partitionBy(numPartitions, partitionFunc), where partitionFunc plays exactly the role getPartition would in Scala.

Structured elaboration

The Scala shape (for reference, matching how the question frames it).

class CustomerPrefixPartitioner(numPartitions: Int) extends Partitioner {
  override def numPartitions: Int = numPartitions
  override def getPartition(key: Any): Int = {
    val customerId = key.toString.split("#")(0)
    (customerId.hashCode & Int.MaxValue) % numPartitions
  }
  override def equals(other: Any): Boolean = other match {
    case p: CustomerPrefixPartitioner => p.numPartitions == numPartitions
    case _ => false
  }
}
// usage: rdd.partitionBy(new CustomerPrefixPartitioner(numPartitions))

getPartition receives each record's key and returns which of numPartitions buckets it belongs to; overriding equals matters because Spark checks whether two RDDs share the SAME partitioner (by equality, not just by class) to decide whether an operation between them can skip a shuffle, so a Partitioner without a correct equals silently loses that optimization even when the partitioning logic is otherwise identical.

Why the key needs to be the PREFIX, not the whole composite key. A naive hash(fullKey) % numPartitions on a key like "cust1#purchase" would place "cust1#click" and "cust1#purchase" in DIFFERENT partitions (different full strings hash differently), defeating the whole purpose; extracting just the customerId portion before hashing is what actually achieves colocation of everything sharing that customer.

Since a Scala compiler was not available in this sandbox, I verified the identical mechanism via PySpark's supported equivalent, rdd.partitionBy(numPartitions, partitionFunc), where partitionFunc is an ordinary Python callable Spark invokes per key, functionally identical to Scala's getPartition (both are called by Spark's shuffle-write path to decide each record's target partition).

Worked example

import zlib

NUM_PARTITIONS = 4

def customer_prefix_partition(key):
    # zlib.crc32, not Python's builtin hash(): builtin string hashing is
    # randomized per-process (PYTHONHASHSEED) unless explicitly disabled,
    # and a partition function shipped to independent executor processes
    # needs the SAME hash value everywhere it runs, or colocation breaks
    # silently and non-deterministically between runs.
    customer_id = key.split("#")[0]
    return zlib.crc32(customer_id.encode("utf-8")) % NUM_PARTITIONS

data = [
    ("cust1#click", 1), ("cust1#purchase", 2), ("cust1#view", 3),
    ("cust2#click", 4), ("cust2#purchase", 5),
    ("cust3#click", 6),
    ("cust4#click", 7), ("cust4#purchase", 8), ("cust4#view", 9), ("cust4#refund", 10),
]
rdd = sc.parallelize(data, numSlices=6)  # start scattered across 6 partitions
partitioned = rdd.partitionBy(NUM_PARTITIONS, customer_prefix_partition)

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

Partition indices used per customer: {'cust4': {0}, 'cust2': {1}, 'cust1': {3}, 'cust3': {3}}
Every customer's records land in exactly ONE partition: True

Per-customer event counts, computed via mapPartitions with NO further shuffle:
('cust1', 3)
('cust2', 2)
('cust3', 1)
('cust4', 4)
Matches expected counts: True

Every one of cust1's 3 records, cust2's 2, cust3's 1, and cust4's 4 landed in a single partition each (note cust1 and cust3 happened to hash to the same partition, 3, which is expected and harmless: colocating DIFFERENT customers together is fine, the property that matters is that no customer's records are SPLIT across partitions). The final step computed each customer's event count via mapPartitions alone, no further shuffle needed, confirming the practical payoff: a downstream per-customer aggregation or join runs entirely within already-local data.

Trade-offs and pitfalls

• Common mistake (the one this exercise deliberately surfaces): using Python's builtin hash() on a string key inside a partition function. Since Python 3.3, string hashing is randomized per process by default (PYTHONHASHSEED) as a security mitigation; a partition function using hash() can compute DIFFERENT bucket assignments for the identical key depending on which process (driver vs. a given executor, or even the same executor restarted) evaluates it, silently breaking colocation in a way that may not be obvious from output alone, only from inconsistent downstream results. zlib.crc32 or hashlib-based hashing gives a hash value that is identical across every process, every run.
• Uneven bucket distribution is a real risk with a small number of distinct prefixes and a much larger number of partitions, exactly as in the worked example (only 4 distinct customers across 4 partitions, so an empty or lightly-loaded partition is likely by chance); a real deployment with a large, well-distributed customer ID space and a reasonably-sized numPartitions relative to that cardinality avoids this, but a small-key-space partitioner should be validated against the ACTUAL key distribution, not assumed even.
• Scala's equals override matters in practice, not just in theory. Two RDDs partitioned by logically-identical but separately-INSTANTIATED CustomPartitioner objects will not be recognized as co-partitioned by Spark's join planner unless equals compares by configuration (here, numPartitions) rather than by object identity, which silently costs an unnecessary shuffle that the whole exercise was meant to avoid.
• This custom partitioning strategy only helps operations that are explicitly partition-aware afterward (mapPartitions-based logic, or a join/cogroup against another RDD sharing the SAME partitioner); a subsequent DataFrame-API operation or an operation that does not check the underlying partitioner will not automatically benefit, since the DataFrame API generally manages its own partitioning through Catalyst rather than inheriting a custom RDD-level partitioner.

Direct answer

MEMORY_ONLY gives the fastest repeated access (no deserialization, no disk I/O) but risks silent partial-recompute if the data does not fully fit; MEMORY_AND_DISK guarantees the full dataset stays available (spilling the overflow to disk rather than dropping it) at some access-latency cost for the spilled portion; DISK_ONLY sacrifices in-memory speed entirely, appropriate specifically when the cached data is too large for memory to matter at all or when memory needs to be reserved for something else. For ML (machine learning) feature pipelines specifically, the right choice tracks how many times a specific feature DataFrame gets REUSED across the pipeline (a single-pass feature transform rarely benefits from caching at all; a feature set reused across many training iterations or multiple downstream model variants benefits substantially).

Structured elaboration

MEMORY_ONLY. Deserialized objects held entirely in-memory; if the dataset does not fit, Spark caches what fits and RECOMPUTES the rest from lineage on each subsequent access needing it (this specific silent-recompute risk and its interaction with serialized storage levels in more depth). For ML feature pipelines: appropriate when the feature set is confidently known to fit in available executor memory and is accessed repeatedly (e.g., cross-validation folds re-reading the same feature set many times), where the speed advantage compounds across many accesses.

MEMORY_AND_DISK. The full dataset is guaranteed to be retrievable without recomputation: whatever does not fit in memory spills to LOCAL disk instead of being dropped. For ML feature pipelines: the safer DEFAULT choice when the feature set's exact size relative to available memory is uncertain (a common situation, since feature engineering often runs before the team has precisely profiled the resulting DataFrame's memory footprint), since it avoids the worst-case surprise of MEMORY_ONLY's silent partial recompute.

DISK_ONLY. No in-memory retention at all; every access reads from local disk. For ML feature pipelines: appropriate when the feature DataFrame is far too large to meaningfully fit in available memory (caching it would spend memory budget for negligible in-memory-cache-hit benefit) but STILL benefits from being materialized once rather than recomputed from a long upstream lineage on every access; essentially, this trades "fast repeated access" for "guaranteed NOT to recompute from source," useful when the source recomputation is itself expensive (a complex upstream join/aggregation chain) even though disk access itself is not fast.

Eviction. Under memory pressure, Spark evicts cached blocks using an LRU (least-recently-used) policy WITHIN the storage-memory pool; a feature pipeline caching MULTIPLE DataFrames concurrently (a common pattern: raw features cached, then several DERIVED feature sets each also cached) can have an EARLIER-cached DataFrame evicted to make room for a LATER one, silently reintroducing recomputation cost for the evicted DataFrame the next time it is needed, a subtle production gotcha worth monitoring via the Storage tab's "Fraction Cached" rather than assumed away.

Serialization. Adding _SER to any of these levels trades CPU (deserialization cost per access) for memory footprint; for ML feature pipelines specifically, where features are frequently NUMERIC (well-suited to compact serialized/columnar representation) and accessed many times across training iterations, the serialized variants are often a good default, since the deserialization cost is usually small relative to the actual model-training computation that follows each access.

GC and shuffle impact. Fewer, more compact objects (serialized storage) reduce GC pressure; a feature pipeline running ALONGSIDE heavy shuffle-based joins/aggregations (common in feature engineering, joining multiple source tables) competes for the SAME unified memory pool between caching and shuffle/execution needs, so an oversized caching footprint can starve execution memory for the surrounding pipeline's shuffles, a trade-off spark.memory.storageFraction tuning directly governs.

Worked example

A feature pipeline computing a 40 GB feature DataFrame (after joining and transforming several source tables) that is then used for 5-fold cross-validation, meaning the SAME feature DataFrame is read 5 separate times by 5 separate training runs, on a cluster with 60 GB of aggregate storage memory available.

MEMORY_ONLY: the 40 GB fits comfortably in 60 GB of storage memory; all 5 cross-validation folds read from memory, no recomputation, the fastest option here since the data genuinely fits.

If the feature set instead grew to 80 GB (exceeding the 60 GB storage-memory budget): MEMORY_ONLY would cache roughly 75% and recompute the remaining ~25% on EVERY one of the 5 folds' accesses, a real, repeated cost (this general pattern); MEMORY_AND_DISK instead spills the overflow to disk once, and all 5 folds read the full 80 GB from either memory or disk, never recomputing, a clear win specifically because this data is reused 5 TIMES (the recompute-avoidance benefit compounds across every reuse, unlike a single-pass job where the choice matters far less).

Trade-offs and pitfalls

• Common mistake: caching a feature DataFrame reflexively because "caching is good practice," without confirming it is actually reused more than once; for a single-pass feature transform feeding directly into ONE training run, caching adds overhead (the caching operation itself, and storage-memory pressure on the rest of the pipeline) for no reuse benefit at all.
• Common mistake: caching MULTIPLE large feature DataFrames concurrently without checking aggregate storage-memory pressure, then being surprised an earlier-cached DataFrame silently evicts and recomputes; the Storage tab's Fraction Cached metric, checked explicitly, is the reliable way to confirm what is ACTUALLY staying cached versus quietly falling back to recomputation.
• Iterative-algorithm reuse versus one-time ETL, the concrete contrast this question's own content depends on: a one-time ETL transform that reads a DataFrame ONCE and writes output gets essentially no benefit from caching that DataFrame at all (there is nothing to REUSE); an iterative algorithm (cross-validation, as in the worked example, or a genuinely iterative training loop reading the same feature set across many epochs) gets a benefit that compounds with EACH additional reuse, which is precisely why "is this dataset accessed more than once" is the single most important question to ask before choosing ANY caching strategy, more important than which SPECIFIC storage level to pick.
• DISK_ONLY is easy to dismiss as "obviously worse" than any memory-involving level, but for a feature set too large to meaningfully benefit from partial in-memory caching, it can be the more HONEST choice: it does not compete for storage memory that the surrounding pipeline's shuffles need, and it still avoids expensive upstream recomputation, a real, if less flashy, benefit.

Direct answer

Moving from YARN to Kubernetes changes where shuffle files physically live and who is responsible for keeping them available: YARN's Node Manager can host Spark's external shuffle service on stable, long-lived cluster nodes, while a Kubernetes pod is ephemeral by design, so shuffle files written to a pod's local (ephemeral) storage disappear the moment that pod is evicted, which breaks dynamic allocation's core assumption that removing an executor does not lose its shuffle output. Getting this right on Kubernetes means either accepting that limitation (disable aggressive dynamic allocation, or use Kubernetes' own executor-removal-aware shuffle tracking) or provisioning persistent volumes specifically for shuffle data.

Structured elaboration

Executor lifecycle differences. On YARN, an executor is a container managed by the NodeManager on a relatively stable physical or VM host; the same host typically persists for the cluster's operational lifetime. On Kubernetes, each executor is its own pod, scheduled onto whichever node the Kubernetes scheduler picks, and pods are treated as disposable: a node drain, an autoscaler scale-down event, or a simple pod eviction under resource pressure can remove an executor pod with much less operational ceremony than decommissioning a YARN NodeManager host.

Shuffle file storage and the external shuffle service problem. Spark's classic external shuffle service runs as a long-lived, cluster-manager-integrated daemon (one per NodeManager on YARN) that keeps serving shuffle files even after the executor that WROTE them has been removed, which is what makes dynamic allocation safe on YARN: an idle executor can be reclaimed without losing in-flight shuffle data other stages still need to read. Kubernetes does not have an equivalent long-lived, node-scoped daemon by default; Spark's Kubernetes shuffle service story has historically lagged YARN's, and the safer default on Kubernetes has been to disable dynamic allocation's more aggressive executor removal, or to rely on Spark's own shuffle-tracking-aware dynamic allocation (spark.dynamicAllocation.shuffleTracking.enabled), which keeps an executor alive until its shuffle data is no longer needed by any active stage, rather than removing it purely on idle timeout.

Persistent vs ephemeral volumes for shuffle. By default, Spark-on-Kubernetes executor pods write shuffle files to the pod's own ephemeral local storage, which is destroyed with the pod. Configuring a persistent volume (a Kubernetes PersistentVolumeClaim mounted for shuffle scratch space, via spark.kubernetes.driver.volumes.*/spark.kubernetes.executor.volumes.* configuration) keeps shuffle data on storage that can outlive the specific pod, though this only helps if the storage is also accessible to WHATEVER pod ends up needing to read it next, which usually means a node-local persistent volume tied to the node (not the pod) plus node-affinity scheduling, or a separately deployed shuffle-service component designed for Kubernetes (community and vendor solutions exist for this).

Tuning steps for stability on Kubernetes:

1. Size executor pod requests/limits to leave headroom matching the same overhead reasoning used on YARN (spark.executor.memoryOverhead), since a pod hitting its Kubernetes memory limit is OOM-killed by the kubelet just as harshly as YARN kills a container over its limit.
2. If dynamic allocation is required, enable spark.dynamicAllocation.shuffleTracking.enabled so executors holding needed shuffle data are not reclaimed mid-stage.
3. Use node-local storage for shuffle scratch space (emptyDir backed by a fast local disk, or a hostPath-style volume) rather than network-backed persistent volumes for the scratch data itself, since shuffle I/O is latency-sensitive and a network volume adds a hop the local NodeManager-hosted shuffle service on YARN did not have.
4. Set Pod Disruption Budgets and appropriate terminationGracePeriodSeconds so the Kubernetes scheduler does not evict executor pods mid-shuffle-write during routine cluster maintenance.
5. Pin resource requests close to limits (avoid Kubernetes "burstable" QoS where possible for executor pods) so the scheduler's bin-packing does not co-locate more executors on a node than its actual disk/network capacity supports, which is easy to do accidentally on Kubernetes since the scheduler is not aware of Spark's per-executor shuffle I/O pattern the way a Hadoop-aware YARN deployment tends to be tuned to be.

Worked example

A batch ETL job with a 50 GB shuffle stage, migrating from a YARN cluster (external shuffle service on stable NodeManager hosts, dynamic allocation freely scaling executors down between stages) to Kubernetes.

On YARN: dynamic allocation removes idle executors between the shuffle-write stage and the next stage that reads it; the external shuffle service on each NodeManager keeps serving the 50 GB of shuffle files to the next stage's tasks regardless of which specific executors are currently running, so aggressive scale-down between stages is safe and saves real compute cost during the gap.

On Kubernetes without shuffle tracking: the same aggressive scale-down would remove executor pods whose LOCAL ephemeral storage still holds unread shuffle output, and the next stage's tasks that need to fetch that data get FetchFailed errors (their intended source is gone) and Spark has to recompute the lost shuffle output from lineage, which costs the entire upstream stage's work again, not just the removed executor's share. With spark.dynamicAllocation.shuffleTracking.enabled=true, Spark tracks which executors hold shuffle data still needed by an active or pending stage and defers their removal until that data is consumed, trading some of the aggressive cost savings YARN's external-shuffle-service model gave for granted for correctness on Kubernetes' more disposable-pod model.

Trade-offs and pitfalls

• Common mistake: enabling dynamic allocation on Kubernetes with the same aggressive idle-timeout settings used on YARN, assuming the external-shuffle-service safety net exists identically; without shuffle tracking (or a real Kubernetes-native shuffle service) enabled, this reliably causes recomputation storms under normal scale-down.
• HDFS access from Kubernetes pods, if migrating off a Hadoop-colocated cluster, needs explicit attention: on YARN, executors often ran on the same physical hosts as HDFS DataNodes, giving natural data locality; Kubernetes pods are typically on separate infrastructure from any legacy HDFS cluster, so reads become remote-network reads regardless of locality settings, and any native Hadoop client libraries or Kerberos configuration baked into the old YARN nodes need to be explicitly rebuilt into the Spark-on-Kubernetes container image and its security configuration (keytabs, delegation tokens) rather than assumed to be present.
• Migration is not purely a shuffle-mechanics change. Driver placement (a Kubernetes pod instead of a YARN ApplicationMaster container), logging/monitoring integration, and image-based dependency management (Kubernetes needs a container image with the right Spark/Python/library versions baked in, versus YARN's shared cluster classpath) all change simultaneously, so a migration plan needs to validate all of these together, not just the shuffle-specific settings above in isolation.

Direct answer

repartition always triggers a full shuffle and can both increase and decrease the number of partitions with the result evenly rebalanced; coalesce avoids a shuffle by default and can only decrease the partition count, by merging existing partitions together without redistributing their contents. Use coalesce when reducing partitions and roughly even sizes are already likely (for example, before a final write, to avoid many small output files); use repartition when you need to increase partitions, need the result evenly balanced regardless of current skew, or need to repartition BY a specific column for a downstream join or aggregation.

Structured elaboration

Cost mechanics. repartition(n) (or repartition(n, col)) always shuffles: every row is reassigned to one of n new partitions via hash (or round-robin, if no column is given) partitioning, guaranteeing a roughly even row count per partition regardless of how skewed the input was. coalesce(n) (with the default shuffle=False) instead merges existing partitions into n groups without moving individual rows between partitions; a coalesce from 200 to 50 partitions, for instance, might simply group every 4 adjacent partitions together, so if the original 200 partitions were unevenly sized, the resulting 50 are unevenly sized in the same proportions, just coarser. This is why coalesce is cheap (no network shuffle) but does not fix imbalance, only reduces partition COUNT.

When coalesce(n, shuffle=True) is the right choice. Passing shuffle=True to coalesce makes it perform an actual shuffle, which sounds like it defeats the purpose, but it is genuinely useful in one specific case: reducing partition count from a much LARGER number to a much SMALLER one when the source partitions are themselves imbalanced, and you need the reduced set to be balanced (not just fewer, but evenly sized). A default coalesce from 2,000 skewed partitions down to 20 could produce a few enormous merged partitions and several nearly-empty ones (since the merge groups adjacent partitions without inspecting their sizes); coalesce(20, shuffle=True) pays the shuffle cost but guarantees evenness, functionally converging on what repartition(20) would do, though the two differ slightly in how they map data into the new partitions internally.

Worked example

A pipeline reads 2,000 small Parquet files (a small-files-problem source), so the initial DataFrame has roughly 2,000 partitions, and needs to write output as roughly 20 well-sized files.

# Cheap: no shuffle, just merges the 2000 partitions down to 20 groups.
# Fine here because the 2000 input partitions are already close to evenly
# sized (they came from similarly-sized small files).
df.coalesce(20).write.parquet("s3://bucket/output/")

Before this final coalesce-and-write, the pipeline computed a heavy aggregation earlier that left the data in 500 shuffle partitions with real skew (one key holds 40% of rows, matching the kind of imbalance a groupBy on a skewed key produces). Writing directly from THAT stage with a plain coalesce(20) would merge the skewed partitions as-is, producing a handful of huge output files and many tiny ones. The fix is repartition(20) (or coalesce(20, shuffle=True)) at that specific point in the pipeline, paying the shuffle cost once to guarantee 20 evenly-sized output files, then coalescing (cheaply, since the data is now already balanced) if any further downstream reduction is needed.

Downstream data locality for ML training. For an ML (machine learning) training job that reads the written Parquet output as its input, partition count and evenness directly affect training throughput: too few, oversized partitions mean each training-data-loading task handles a large chunk serially and the cluster's read parallelism is underused; a handful of oversized partitions from a naive coalesce after a skewed aggregation would show up downstream as some data-loading tasks taking far longer than others, exactly the straggler pattern skew produces elsewhere, just relocated to the training-read stage instead of the aggregation stage.

Trade-offs and pitfalls

• Common mistake: calling coalesce to reduce partitions after a wide (shuffle-producing) transformation and expecting it to also fix any skew that transformation left behind. Coalesce only merges existing partitions; if they were unbalanced going in, they remain unbalanced (just fewer and larger) coming out.
• Common mistake: using repartition reflexively everywhere partition count needs to change, even when reducing count and the data is already balanced, paying an unnecessary shuffle that coalesce would have avoided.
• A self-tuning approach worth knowing: rather than a fixed partition count, some pipelines sample the DataFrame's estimated size (via df.rdd.getNumPartitions() combined with an estimate of average partition byte size, or Catalyst's own size statistics when available) to compute a recommended N for a final coalesce or repartition, targeting a fixed per-file size (for example 128 to 256 MB) rather than a hardcoded partition count; this generalizes better across datasets of varying size than a magic-number coalesce(20) chosen for one particular run.
• coalesce reducing partition count also reduces parallelism for any transformation chained immediately after it, not just for the final write; calling coalesce(1) early in a pipeline to "clean up" partition count before more transformations follow is a common performance bug, since it forces all subsequent work for that stage onto effectively 1 task.

Direct answer

Distinguish storage-memory pressure from execution-memory pressure FIRST, since they have different root causes and different fixes: check the Spark UI's Storage tab (is a cached dataset only partially fitting, forcing repeated recomputation) versus the Executors tab's GC-time and the Stages tab's spill metrics (is the actual COMPUTATION, not caching, running out of working memory). GC pauses and intermittent OOMs together point most commonly at execution memory genuinely undersized for the workload's actual per-task working set, often triggered by a specific stage (a wide aggregation, a large broadcast, a memory-hungry UDF) rather than a uniform, job-wide problem.

Structured elaboration

Step 1: logs and metrics to inspect, in order.

1. Executors tab, GC Time column, across all executors: elevated GC time (a large fraction of total task time) confirms genuine memory pressure exists, distinguishing this from an unrelated slowness cause entirely.
2. Stages tab, per-stage Spill (Memory)/Spill (Disk) columns: identifies WHICH SPECIFIC STAGE is causing the pressure, since "intermittent" often means one particular stage (not every stage) is the actual source, and pinning down the specific stage narrows the next investigation substantially.
3. Executor logs for any OOM-killed executor, checking for the specific exception (a java.lang.OutOfMemoryError: Java heap space from within Spark's own execution versus a cluster-manager-level container-memory-limit kill, distinguishing these): a heap-space OOM points at Spark's own JVM heap being undersized for its workload; a container-limit kill points at memoryOverhead being undersized relative to what the JVM plus any off-heap/native usage (Python UDF worker processes, native library memory) actually needs.

Step 2: storage versus execution memory. Spark's unified memory model (spark.memory.fraction, spark.memory.storageFraction) shares one memory region between STORAGE (cached data) and EXECUTION (working memory for shuffles, joins, aggregations, sorts) with a degree of mutual borrowing; a job caching a large dataset AND running memory-intensive execution work concurrently can have EITHER side of that shared pool as the actual pressure source. Check the Storage tab's "Fraction Cached" for the specific cached dataset(s) in the job: a cached dataset that is NOT fully cached (forcing partial recomputation from lineage on every access) can indirectly cause what LOOKS like intermittent OOM/GC pressure, since the recomputation path itself may be what is memory-hungry, not the caching itself.

Step 3: possible fixes, matched to the confirmed cause.

• If execution memory is undersized for a specific stage's actual working set (confirmed via that stage's spill metrics): increase spark.executor.memory and/or spark.memory.fraction, or reduce the per-task working-set size directly by increasing shuffle-partition count for that specific stage (smaller partitions, the heuristic).
• If serialization is contributing (a job using Java serialization instead of Kryo, the trade-off): switching to Kryo reduces both the memory footprint of cached/shuffled objects AND GC pressure (fewer, more compact objects for the GC to trace), often a meaningful lever even when the root sizing is otherwise reasonable.
• If a specific caching choice is the culprit (a large dataset cached MEMORY_ONLY that does not fully fit): switching to a serialized or disk-backed storage level directly addresses the memory footprint without needing to grow the cluster.
• If a specific UDF or code path is unusually memory-hungry (a Python UDF materializing a large intermediate structure per row, or a custom aggregation with an unbounded accumulator): the fix is in the CODE itself, not a configuration change; this is why Step 1's stage-level narrowing matters, since a configuration-only response to a code-level problem treats the symptom without fixing the actual cause, and the underlying code issue will likely resurface under slightly different conditions.

Worked example

A production job shows GC-time spikes and occasional executor OOM roughly once every several runs, not every run, hence "intermittent." Checking the Executors tab confirms elevated GC time correlates specifically with runs where a particular UPSTREAM data source had an unusually large day's volume (visible by cross-referencing job-level input-row-count metadata against the runs that showed GC/OOM symptoms). Checking the Stages tab for those specific runs shows the elevated spill concentrated in ONE stage: a groupBy aggregating a JSON-parsed nested structure per key, where the per-key accumulator holds a growing LIST of values (a collect_list-style aggregation) rather than a fixed-size running total; on a normal-volume day, this list stays small enough to fit comfortably, but on a high-volume day, some keys' accumulated lists grow large enough to individually stress a single task's memory, a workload characteristic that DOES scale with input volume in a way a simple running sum/count would not. Root cause: an aggregation whose per-group memory footprint is NOT bounded independent of input volume (unlike the "number of groups drives memory, not row count" pattern that holds for simple aggregations, that general case, which explicitly does not apply here because THIS aggregate function's state genuinely grows with input). Fix: either accept the larger per-key memory footprint and size spark.executor.memory for the WORST-CASE (highest-volume) day rather than a typical day, or, if the full per-key list is not actually needed downstream, redesign the aggregation to compute only what is genuinely needed (a bounded summary statistic instead of the full list) to avoid the unbounded-per-group memory growth entirely.

Trade-offs and pitfalls

• Common mistake: responding to intermittent GC/OOM by uniformly increasing executor memory for the whole cluster/job, without first narrowing down WHICH stage and WHY; this often "fixes" the symptom for a while (more headroom absorbs the problem until volume grows again) while leaving an unbounded-per-group aggregation (as in the worked example) as a ticking time bomb that will resurface once volume grows past whatever headroom was added.
• Common mistake: treating "intermittent" as inherently mysterious or hard to diagnose; as the worked example shows, intermittent symptoms are frequently explained by a specific, identifiable VARYING FACTOR (data volume on a specific day, a specific upstream source's occasional burstiness) rather than true randomness, and cross-referencing the symptom's occurrence against such a factor is usually the fastest path to root cause.
• This general troubleshooting shape (check GC/spill metrics, distinguish storage from execution memory, look for a code-level cause behind a configuration-level symptom) generalizes beyond THIS specific scenario: it is the same diagnostic discipline to skew, straggler tasks, and shuffle failures, treating "add more resources" as a last resort after narrowing down the actual mechanism, not a first response.
• Distinguishing a heap-space OOM from a container-memory-limit kill matters for which config to change: increasing spark.executor.memory (the JVM heap) does nothing for a container-limit kill caused by insufficient memoryOverhead (off-heap/native usage), and increasing memoryOverhead does nothing for a genuine JVM heap-space OOM; applying the wrong one of the two fixes can look like "it didn't help" when actually the WRONG lever was pulled for the confirmed cause.