Distributed Data Processing with Spark and Hadoop Questions

Direct answer

Set the watermark to the full 48-hour late-arrival tolerance (withWatermark("event_time", "48 hours")), size the state store to hold every open session across that entire 48-hour window (not just the current moment), and route checkpointing to durable, versioned storage since a state store this large makes restart-from-scratch prohibitively expensive. Backfilling historical data safely means treating the backfill as a SEPARATE batch job writing into the same target table, never replaying old data through the LIVE streaming query's state, since replaying old events into an active watermark-bounded query either gets silently dropped (older than the current watermark) or corrupts the state store's notion of "current."

Structured elaboration

Watermark setting. withWatermark("event_time", "48 hours") tells Structured Streaming to keep a session/window's state open until the maximum observed event time advances 48 hours past that session's last event, meaning genuinely late events within that window are still correctly incorporated; anything later than 48 hours is dropped by design. This threshold is chosen entirely by the business requirement stated in the question (events can be up to 48 hours late), not by any Spark-internal default; a shorter watermark would silently drop legitimate late events the business has explicitly said to expect.

State retention and sizing state store resources. A 48-hour watermark means the state store must hold every open session (or window) whose data could still legitimately arrive, for up to 48 hours of wall-clock/event-time span, which for a high-volume session-metrics job can be a genuinely large amount of state (potentially many GB to TB, depending on session volume and per-session state size). This directly drives infrastructure sizing: executor memory/disk needs to accommodate the state store's expected size (Spark's default in-memory state store keeps state in executor memory backed by periodic snapshots to the checkpoint location; RocksDB-backed state store, spark.sql.streaming.stateStore.providerClass set to the RocksDB provider, moves the bulk of state to local disk with only a working set in memory, which matters specifically for state sizes that would not comfortably fit in-memory for a 48-hour retention window).

Checkpointing location. Given the state size implied by 48 hours of retention, checkpoint storage needs both the durability (a real, replicated store, not local-only) AND the throughput to handle periodic full or incremental state snapshots without becoming the pipeline's bottleneck; this is a case where the earlier general "reliable vs local checkpointing" trade-off resolves clearly toward reliable, high-throughput durable storage, since the state being protected is both large and expensive to rebuild from scratch.

Handling events beyond the watermark. Once an event's timestamp falls outside the 48-hour watermark window, Structured Streaming drops it from the stateful computation by design; there is no built-in mechanism to retroactively "un-drop" it into the live query. If genuinely-too-late events still need to be captured (for audit, or for a slower-but-more-complete reconciliation), the standard pattern is a SEPARATE sink capturing all raw events regardless of lateness (a raw landing table with no watermark logic applied), from which a periodic batch reconciliation job can correct the aggregated output after the fact, entirely outside the live streaming query's state.

Backfilling historical data without corrupting streaming state. Feeding old historical data through the SAME live streaming query is the specific failure mode to avoid: if the historical data's event times are older than the query's current watermark, it is silently dropped (no corruption, but also no effect, wasted work); if it happens to fall WITHIN the current watermark window, it can genuinely corrupt in-progress session/window state by mixing historical replay with live traffic in ways the downstream consumer of the output was not expecting. The safe pattern is a SEPARATE batch job, reading the same historical source, applying the identical session/window logic via the batch DataFrame API (the same code, since Structured Streaming and batch DataFrames share the same API and Catalyst optimizer), and writing to the target table with a clear boundary (a partition, a date range, or an explicit "backfill" marker) so the backfilled output is auditable and does not silently blend with or double-count the live query's output for the same period.

Worked example

A session-metrics pipeline: groupBy(session_window(event_time, "30 minutes"), user_id) with withWatermark("event_time", "48 hours"), writing finalized session summaries to a Delta table.

sessions = (events
    .withWatermark("event_time", "48 hours")
    .groupBy(F.session_window(F.col("event_time"), "30 minutes"), "user_id")
    .agg(F.count("*").alias("event_count"), F.min("event_time").alias("session_start"),
         F.max("event_time").alias("session_end")))

query = (sessions.writeStream
    .outputMode("append")
    .format("delta")
    .option("checkpointLocation", "/checkpoints/session-metrics")  # durable, replicated storage
    .option("path", "/lake/session_metrics")
    .trigger(processingTime="1 minute")
    .start())

A discovered gap: three weeks of historical raw events were never processed by this pipeline (it was deployed after that data already existed). The correct backfill is a batch job:

historical = spark.read.format("delta").load("/lake/raw_events") \
    .filter("event_time between '2026-07-01' and '2026-07-21'")
backfilled_sessions = (historical
    .groupBy(F.session_window(F.col("event_time"), "30 minutes"), "user_id")
    .agg(F.count("*").alias("event_count"), F.min("event_time").alias("session_start"),
         F.max("event_time").alias("session_end"))
    .withColumn("source", F.lit("backfill")))  # explicit marker distinguishing from live output
backfilled_sessions.write.format("delta").mode("append").save("/lake/session_metrics")

Running this as a batch job entirely separate from the live streaming query means the live query's checkpoint, watermark, and state store are never touched by the backfill, and the source="backfill" marker makes the two populations distinguishable for any downstream consumer that needs to know which rows came from which path.

Trade-offs and pitfalls

• The lookup-refresh problem: refreshing a slowly-changing reference/lookup table used INSIDE the streaming query (for example, a user-attributes table joined against the event stream) without triggering a large shuffle on every refresh is a related but distinct concern from state retention; the standard approach is a periodically-refreshed BROADCAST join against the lookup table (re-loaded on a schedule outside the streaming query's own micro-batch loop, using foreachBatch to re-read and re-broadcast the lookup at controlled intervals) rather than re-joining against a large, un-broadcast lookup table on every micro-batch, which would add shuffle cost proportional to lookup table size on every single trigger.
• Common mistake: setting the watermark shorter than the business's actual stated late-arrival tolerance to reduce state size, silently trading away data completeness the business explicitly asked for; if 48 hours of state genuinely does not fit the infrastructure budget, that is a capacity conversation to have explicitly, not a watermark value to quietly shrink.
• Common mistake: replaying historical data through the live query's Kafka (or other) source topic to "backfill" it, assuming the streaming job will just pick it up correctly; this is exactly the state-corruption/silent-drop failure mode described above, and the batch-job backfill pattern exists specifically to avoid it.
• 48 hours of state retention makes the RocksDB-backed state store provider worth strongly considering over the default in-memory provider, since in-memory state at this scale competes directly with executor heap needed for the rest of the computation, while RocksDB moves the bulk of state to local disk with a smaller in-memory working set, a trade-off (some added per-access latency for a much larger sustainable state size) that generally favors RocksDB once state size grows past what comfortably fits in a reasonable fraction of executor heap.

Direct answer

Setting spark.sql.shuffle.partitions too low forces too much data into too few partitions, which causes large per-task memory pressure, disk spills, and long straggler tasks that leave most of the cluster idle; setting it too high creates thousands of tiny tasks whose per-task scheduling overhead (a few milliseconds of driver bookkeeping and executor launch overhead each) starts to dominate the actual work, and it multiplies the number of small output files. A reasonable starting value is derived from the shuffle stage's actual data size, not left at Spark's fixed default of 200.

Structured elaboration

Too low. Fewer partitions means each partition holds more data. If a partition's data (plus the intermediate structures the shuffle build needs) exceeds what fits comfortably in that task's execution memory, the task spills to disk, and disk I/O is far slower than in-memory processing; in the worst case a single oversized partition causes an executor OOM (out of memory) and the task, then the stage, fails and retries, wasting the work already done by every other task in that stage. Too-low partition counts also directly limit parallelism: if you have 300 available cores but only 50 shuffle partitions, at most 50 cores can be doing shuffle-stage work at once, and 250 sit idle.

Too high. Each task carries fixed overhead: driver-side task scheduling, serialization of the task closure, executor JVM bookkeeping to track the task, and (for writes) one output file per task per partition column value. At tens of thousands of partitions, this overhead, while small per task, adds up to a meaningful fraction of total stage time, and the output-file explosion becomes its own downstream problem (the small-files problem: more metadata for the catalog to track, more S3/HDFS GET operations for the next job that reads the output, lower per-file read throughput).

Rule-of-thumb formula. Target roughly 128 to 200 MB of shuffle data per partition (matching common HDFS/Parquet block sizes, which keeps individual task I/O well-sized), and set:

Cross-check this against available parallelism: the result should be at least $2$ to $3\times$ the total number of executor cores, so the cluster gets multiple waves of tasks rather than one wave that leaves cores idle at the tail; take the max of the size-based and parallelism-based numbers.

How skew and cluster size change the choice. If the shuffle key is skewed, no single partition-count choice fixes the imbalance, since one key's rows still land in one partition regardless of the total count; the formula above sizes for the AVERAGE partition, and a skewed job needs a separate mitigation (salting, broadcast join, or Adaptive Query Execution's runtime skew handling) on top of a reasonable base partition count. On a small cluster, the parallelism floor (2 to 3x cores) is usually the binding constraint since there is little total data per partition to begin with; on a large cluster processing a large shuffle, the size-based target usually dominates.

Worked example

A shuffle stage moving 40 GB of intermediate data, on a cluster with 200 total executor cores.

By size, at a 128 MB target: $40 \times 1024 \text{ MB} / 128 \text{ MB} = 320$ partitions.
By parallelism, at 2 to 3x cores: $400$ to $600$ partitions.
Take the max: start at roughly 500 to 600 partitions, closer to the parallelism floor since it exceeds the size-based number here.

Compare this to leaving the default spark.sql.shuffle.partitions=200 unchanged: 200 partitions for 40 GB gives $40 \times 1024 / 200 = 204.8$ MB per partition, which is not disastrously oversized, but is well below the 500 to 600 partition parallelism floor for a 200-core cluster: with only 200 partitions and 200 cores, the stage runs in a single wave with essentially no headroom to absorb a slow task or minor skew, whereas 500 to 600 partitions gives 2 to 3 waves, so a straggler in wave 1 does not stall the entire stage the way it would with exactly one wave.

Trade-offs and pitfalls

• The default (200) is a fixed constant, not adaptive to job size. It was reasonable for the mid-size jobs common when it was chosen, but it is frequently wrong at both ends: too many for a small aggregation over a few GB, too few for a shuffle over hundreds of GB. Adaptive Query Execution's shuffle-partition coalescing (default on since Spark 3.2) mitigates the too-many-partitions case automatically at runtime by merging small adjacent partitions post-hoc, which reduces the cost of overshooting the partition count, but does not fix a too-low starting value causing spills.
• Common mistake: picking one partition count for the whole job. A pipeline with multiple shuffle stages of very different sizes (say a 40 GB shuffle followed by a 2 GB shuffle) benefits from setting spark.sql.shuffle.partitions differently around each stage (it can be changed per SparkSession.conf.set call between operations) rather than one global value sized for the largest stage, which would over-partition the smaller one.
• Common mistake: tuning partition count without checking AQE's effect first. With AQE enabled, the CONFIGURED shuffle-partition count becomes a starting point that AQE may coalesce down at runtime based on actual observed partition sizes; verify the EFFECTIVE partition count in the Spark UI's SQL tab rather than assuming the configured value is what ran.

Direct answer

Compute a stable, order-independent checksum per row (hashing an explicitly-ordered, explicitly-cast concatenation of the reconciled columns, never the row's raw physical bytes or column order) on both source and reporting tables, then full-outer-join on the business key and classify each row as MATCH, VALUE_MISMATCH, MISSING_IN_REPORTING, or EXTRA_IN_REPORTING by comparing checksums and null-presence. Broadcast the smaller side of the join if one side fits comfortably in executor memory, since reconciliation jobs frequently compare a large source table against a much smaller reporting rollup.

Structured elaboration

Computing a stable checksum across possible schema changes. Hashing raw row bytes (hash(*) over all columns positionally) breaks the moment column order changes or a new column is added to one side but not the other, silently corrupting the comparison rather than failing loudly. The robust approach explicitly lists the columns being reconciled, casts each to string (so a 998.10 double and a "998.10" string-typed source column, if that mismatch existed, would be compared consistently rather than by incompatible native types), coalesces NULL to an explicit sentinel string (so a missing value hashes differently from a present-but-empty one, and two independently-NULL columns still compare as equal to each other rather than each hashing as an unpredictable NULL-propagated result), and concatenates with an explicit delimiter before hashing (sha2(..., 256), cryptographically collision-resistant enough that an accidental false-match between two genuinely different rows is not a practical concern at reconciliation scale).

Partitioning and joins. The reconciliation join needs every row for a given key from BOTH sides to be compared, which is exactly a join's job; the choice between a shuffle (sort-merge) join and a broadcast join follows the same size-based reasoning as any other Spark join: if the reporting side is a monthly rollup (thousands to low millions of rows) and the source is the full underlying transaction table (potentially billions of rows), broadcasting the reporting side avoids shuffling the much larger source table at all.

Performance optimizations for the checksum comparison itself. Computing the checksum is an embarrassingly parallel row-level operation (no shuffle needed for the hash computation itself, only for the subsequent join), so its cost scales linearly with data size and parallelizes cleanly across partitions; the join afterward is where the real cost and tuning decisions (broadcast threshold, shuffle partition count, bucketing both tables on the reconciliation key if this job runs repeatedly on a schedule) live.

Worked example

from pyspark.sql import functions as F

def stable_checksum(df, cols):
    concat_expr = F.concat_ws("|", *[
        F.coalesce(F.col(c).cast("string"), F.lit("<NULL>")) for c in cols
    ])
    return df.withColumn("row_checksum", F.sha2(concat_expr, 256))

checksum_cols = ["user_id", "revenue", "month"]
source_cs = stable_checksum(source, checksum_cols).select("user_id", "revenue", "row_checksum")
reporting_cs = stable_checksum(reporting, checksum_cols).select("user_id", "revenue", "row_checksum")

diff = (source_cs.alias("s")
        .join(F.broadcast(reporting_cs).alias("r"), on="user_id", how="full_outer")
        .select(
            F.coalesce(F.col("s.user_id"), F.col("r.user_id")).alias("user_id"),
            F.col("s.revenue").alias("source_revenue"),
            F.col("r.revenue").alias("reporting_revenue"),
            F.col("s.row_checksum").alias("source_checksum"),
            F.col("r.row_checksum").alias("reporting_checksum"))
        .withColumn("delta_type",
            F.when(F.col("source_checksum").isNull(), F.lit("EXTRA_IN_REPORTING"))
             .when(F.col("reporting_checksum").isNull(), F.lit("MISSING_IN_REPORTING"))
             .when(F.col("source_checksum") != F.col("reporting_checksum"), F.lit("VALUE_MISMATCH"))
             .otherwise(F.lit("MATCH"))))

diff.filter(F.col("delta_type") != "MATCH").orderBy("user_id").show()

Test data: source has u1..u4; reporting has u1,u2,u3 (with u3's revenue corrupted from 998.10 to 999.10), u5 (present only in reporting), and is missing u4 entirely.

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

+-------+--------------+-----------------+--------------------+--------------------+--------------------+
|user_id|source_revenue|reporting_revenue|     source_checksum|  reporting_checksum|          delta_type|
+-------+--------------+-----------------+--------------------+--------------------+--------------------+
|     u3|         998.1|            999.1|289f69332563db015...|b9c7bf2cc180b25e5...|      VALUE_MISMATCH|
|     u4|          10.0|             NULL|46e14cd3c32ad27c5...|                NULL|MISSING_IN_REPORTING|
|     u5|          NULL|              5.0|                NULL|c4ddea5051d7f2c14...|  EXTRA_IN_REPORTING|
+-------+--------------+-----------------+--------------------+--------------------+--------------------+

All three injected discrepancies (a value mismatch, a missing row, and an extra row) were correctly identified and classified, and u1/u2 (identical on both sides) correctly did not appear in the mismatch output at all, confirming the checksum-equality path (the MATCH case) works and does not false-positive.

Trade-offs and pitfalls

• Common mistake: hashing concat(*) without explicit column ordering and NULL handling, which silently breaks the moment a schema evolves; the explicit column list and NULL-sentinel approach above costs a little more code but is the difference between a reconciliation job that fails LOUDLY on a schema mismatch (a column list that no longer matches reality is a code change someone has to make deliberately) versus one that silently produces wrong comparisons.
• Common mistake: joining on the FULL row instead of the business key, which cannot distinguish "this row changed" (the actual, useful signal) from "this row does not exist in the other table"; joining on the key first and comparing checksums SEPARATELY is what makes the four-way classification (match/mismatch/missing/extra) possible at all.
• Floating-point revenue values reconciled via string-cast-then-hash are sensitive to representation differences (998.10 vs 998.1 vs 998.100000001 from a floating-point rounding difference upstream) that are not really data errors; for currency-like values, casting to a fixed-precision decimal type before the string cast (rather than relying on a floating-point double's default string representation) avoids flagging cosmetic representation differences as VALUE_MISMATCH false positives.
• Broadcasting the "smaller" side is only safe if it is ACTUALLY small enough; blindly broadcasting a reporting table that has grown past spark.sql.autoBroadcastJoinThreshold (or past what actually fits comfortably in every executor's memory) risks the same broadcast-join memory risk, and should be validated with real size checks, not assumed from a table's historical size.

Direct answer

TaskNotSerializableException means a closure Spark needs to ship to executors (a lambda or function passed to map/filter/etc.) captured a reference to an object that Java's serialization cannot handle, most commonly a large third-party object (a database connection, an HTTP client, a non-serializable library instance) pulled in accidentally because it was declared OUTSIDE the closure and referenced from inside it. The fix is almost always to keep the non-serializable object OUT of the closure's captured scope entirely: construct it INSIDE the closure (or inside mapPartitions, once per partition) rather than referencing an already-constructed instance from the driver's scope.

Structured elaboration

How closure serialization works. When a transformation like rdd.map(f) is submitted, Spark serializes the function f (including everything it references from its enclosing scope, its "closure") on the driver and ships that serialized closure to each executor, which deserializes and runs it once per record (or, for mapPartitions, once per partition). Any object referenced inside f that lives OUTSIDE f's own local scope, a class-level field, an object declared earlier in the same Scala object/class, or a Python object captured via closure, gets pulled into that serialized bundle whether or not the code's author intended it.

Common sources in Scala. A Spark job written inside a Scala class (not an object) often accidentally captures a reference to the ENTIRE enclosing class instance (this) inside a closure, even if the closure only actually uses one field of that class, because Scala's closure capture semantics capture the whole enclosing object by default unless the specific field is extracted into a local val first. If that enclosing class also holds a non-serializable field (a SparkContext reference, a database connection pool), the WHOLE class becomes part of the closure's captured state, and serialization fails on that unrelated field even though the closure's actual logic never touches it.

Common sources in PySpark. Referencing a large or non-serializable Python object defined at MODULE or DRIVER scope from inside a map/filter lambda (a database connection object, an unpicklable third-party client instance, a large in-memory model object not designed for pickling) causes cloudpickle (PySpark's serializer for Python closures) to attempt to serialize that object as part of shipping the closure, which either fails outright (if the object genuinely cannot be pickled) or succeeds but ships a large, wasteful payload to every task.

Practical fixes, in order of preference.

1. Construct the object INSIDE the closure, so it is created fresh on each executor rather than serialized from the driver: rdd.map(lambda x: SomeClient().call(x)) instead of client = SomeClient(); rdd.map(lambda x: client.call(x)). This avoids serialization entirely, at the cost of re-constructing the object once per RECORD, which is wasteful if construction itself is expensive.
2. Use mapPartitions to construct the object once PER PARTITION, amortizing construction cost across all records in that partition rather than paying it per record: rdd.mapPartitions(lambda partition: (SomeClient().call(x) for x in partition)), the standard pattern for anything with meaningful setup cost (a database connection, a loaded model) that also cannot cross the serialization boundary.
3. Extract only the specific serializable field needed, into a local variable, BEFORE the closure, when the goal is genuinely to reuse a value computed on the driver (not to avoid constructing something expensive): val threshold = config.threshold; rdd.filter(x => x.value > threshold) in Scala, capturing only the primitive threshold value rather than the whole config object.
4. Broadcast the object, if it is read-only, genuinely serializable, and needed identically across all tasks: spark.sparkContext.broadcast(largeButSerializableObject), shipping it once per executor (not once per task) and referencing .value inside the closure.

Worked example

# BROKEN: db_client is a driver-side object; referencing it inside the
# closure forces PySpark to try to pickle it, which fails for most real
# database client objects (they hold open sockets/connections, which are
# not meaningfully serializable).
db_client = SomeDbClient(connection_string)
result = rdd.map(lambda row: db_client.lookup(row.id))  # raises serialization error

# FIXED (per-record construction, simplest, costs one connection per record):
result = rdd.map(lambda row: SomeDbClient(connection_string).lookup(row.id))

# FIXED (per-partition construction via mapPartitions, amortizes setup cost):
def lookup_partition(rows):
    client = SomeDbClient(connection_string)  # built once per partition
    try:
        for row in rows:
            yield client.lookup(row.id)
    finally:
        client.close()

result = rdd.mapPartitions(lookup_partition)

The mapPartitions version is the one actually recommended for a real database client: constructing a fresh connection per RECORD (the simplest fix) is correct but can be prohibitively slow and can exhaust connection limits on the database side at scale, while mapPartitions amortizes the connection setup to once per partition, a substantial difference at any real data volume.

Trade-offs and pitfalls

• Common mistake: fixing this by wrapping the object in a broadcast variable when it is NOT actually safe to share (a stateful connection object, something with a lifecycle tied to a single process); broadcast is appropriate for read-only, genuinely shareable data, not for a connection-holding client, which each executor still needs its OWN independent instance of.
• Common mistake: in Scala specifically, moving code into an object (Scala's singleton) as a blanket fix without understanding WHY it sometimes helps; an object's members can be accessed without capturing an outer class INSTANCE the way a class's members do, which avoids one common capture pattern, but does not fix every serialization issue (a genuinely non-serializable field referenced from inside the object still fails).
• A related, deeper case: a Scala custom-typed Aggregator (org.apache.spark.sql.expressions.Aggregator) needs its own Encoder for its intermediate/output types, and a failure to provide (or correctly derive) that Encoder produces a similar-looking serialization failure specific to how Spark SQL encodes typed intermediate state for a custom aggregation, distinct from but related to the general closure-capture problem: both ultimately trace back to Spark needing to know HOW to serialize something it is being asked to ship across the cluster, whether that is a plain closure or a typed aggregation's accumulator state.
• The per-record-construction fix (option 1) trades correctness for potential performance cost; it is the right first fix to unblock a failing job, but for anything with real construction overhead, migrating to mapPartitions (option 2) is worth the small extra code complexity once the job is otherwise working.

Direct answer

Start with the Stages tab to find which stage is actually slow and why (skew, GC, or shuffle cost show up there directly), then use Executors to check for memory pressure or lost/restarted executors across the cluster, then Storage if caching is involved, and SQL/History Server only when you need the full query plan or a completed job's metrics after the fact. The order matters: Stages tells you WHERE the time went before you spend time guessing WHY.

Structured elaboration

Stages tab, first stop. Open the slow stage and look at the task-level "Summary Metrics" table: task duration distribution (median vs max reveals straggler tasks, a skew signature), shuffle read/write size per task (large variance here also signals skew), and the GC time column (a high GC time relative to task duration signals memory pressure, not skew). This single view answers "is this stage slow because of one bad partition, or is every task uniformly slow?", which determines whether the next investigation should be about data distribution or about resource sizing.

Tasks (within a stage), second. Sort tasks by duration descending and inspect the slowest few individually: their host/executor ID (are the slow tasks concentrated on one executor, suggesting a hardware or node-level issue, or spread across many, suggesting a data issue?), their "Shuffle Spill" columns (spilling while sibling tasks do not is a size-per-partition problem), and whether they show FetchFailed or similar errors (a shuffle-fetch or network problem, not a compute problem).

Executors tab, third. Check per-executor "Storage Memory" and "Task Time (GC Time)" columns across the whole cluster: an executor with disproportionately high GC time is running low on usable heap for its workload; an executor that is missing from the current list but was present earlier (visible in the "Removed Executors" section) indicates it died and was replaced, which is itself a strong signal (OOM kill, or a lost node) worth investigating via that executor's logs.

Storage tab, fourth, only if caching is involved. Confirms whether a cache()/persist() call actually achieved the storage level and fraction requested (the "Fraction Cached" column); a job relying on a cached DataFrame that only partially fit in memory (showing less than 100% cached) will silently recompute the uncached fraction from source on every access, which looks like generic slowness until you check this specific tab.

SQL tab, fifth. For DataFrame/SQL-API jobs, shows the physical query plan with per-operator metrics (rows processed, time spent) directly annotated on the plan graph; useful once you already suspect a specific operator (a particular join or aggregate) and want to confirm it against the actual physical plan Catalyst chose, rather than the plan you assumed it would choose.

History Server, when the job has already finished. The same tabs, but for a completed application; essential when the failure or slowness was observed after the fact (an alert fired, a downstream job noticed late data) rather than while watching a live job, since the live UI's in-memory event data is lost once the driver process exits unless event logging (spark.eventLog.enabled=true) was on.

Worked example

An ML training job that reads and joins two feature tables runs noticeably slower than its usual pattern. Following the order above:

1. Stages tab: one stage (the join stage) shows a median task duration of a few seconds but a maximum task duration an order of magnitude higher, with only 2 to 3 tasks in that range out of hundreds. That shape (most tasks fast, a small number far slower) is the classic skew signature, not a uniform slowdown.
2. Tasks within that stage, sorted by duration: the 2 to 3 slow tasks show shuffle-read sizes far above the median for that stage and non-zero "Shuffle Spill (Disk)" while the fast tasks show none, confirming those specific tasks are handling disproportionately large partitions.
3. Executors tab: GC time is unremarkable across the board, ruling out a general memory-pressure explanation and reinforcing that this is a data-shape (skew) problem, not a sizing problem.
4. Conclusion: the join key is skewed; the next step is a mitigation (salting, broadcast join if one side is small enough, or enabling Adaptive Query Execution's runtime skew handling), not a resource-sizing change, because resource sizing does not fix the fact that one key's rows all land on one task.

Trade-offs and pitfalls

• Common mistake: jumping straight to Executors and concluding "we need bigger machines" from high aggregate GC time without checking Stages first for whether the slowness is concentrated (skew) or genuinely uniform (real memory pressure); the fix for each is different and the wrong one wastes cluster budget without resolving the symptom.
• Common mistake: ignoring the "Removed Executors" section and only looking at currently-live executors; a job that lost and replaced 3 executors mid-run had real failures worth investigating even if the job eventually succeeded, since the retries add real cost and the root cause (an OOM, a bad node) will likely recur.
• The SQL tab's plan view is most useful when you already have a hypothesis (from Stages/Tasks) about which operator is the problem; reading it cold on a complex multi-join query without that narrowing first is a much slower way to find the same answer.
• History Server retention is finite and configurable (spark.history.retainedApplications/spark.eventLog.dir disk usage); for a job you may need to debug after the fact, verify event logging is actually enabled and retained long enough before you need it, not after.