Apache Spark / Advanced: streaming and recoverable operation / Lesson 15
Skew, memory and failure diagnosis
Classify a slow or failed stage using task-level evidence before changing resources or protective settings.
On this page
Lesson goals
- Separate skew, join expansion, JVM pressure and Python-worker memory.
- Evaluate a narrow remedy with correctness and recovery evidence.
The last error is not always the first cause
A stage can fail after an executor disappears, leaving fetch errors in downstream tasks. Increasing retries may reproduce the same failure without repairing its cause. Capture the approved driver and executor evidence, first relevant exception, application attempt and stage timeline. Distinguish a driver collect failure from an executor heap error, a container memory termination and a Python-worker exception. These involve different memory pools and responsibilities. A generic “out of memory” summary loses the information needed to choose a remedy. No log collection or configured cluster inspection is performed by this course.
Compare the task distribution rather than the average
The synthetic metric fragment describes a stage whose maximum task duration is much larger than its median. That suggests a tail worth investigating, not proof of one specific hot key. Compare bytes, records, spill, join output and executor placement for slow tasks. A few enormous rows can differ from one frequent site key, and accidental many-to-many join expansion can imitate skew. Validate the row contract before salting or broadcasting. These numbers are a local diagnostic fixture, not measurements from a Spark application or a capacity recommendation.
{"scope":"synthetic stage fixture","median_task_ms":200,"max_task_ms":6000,"next_evidence":["records","shuffle bytes","spill","join output grain"]}The maximum is30 times the median. That ratio identifies a diagnostic question, not an engine-tested root cause.
A hint changes a strategy rather than the data contract
Broadcasting a genuinely small site dimension can avoid a shuffle strategy for a supported join, but the relation must fit the relevant memory and transfer budgets. A row-count estimate alone can mislead when labels or metadata are large. The fragment proposes a broadcast hint for an existing lab dimension; it does not force a universal successful plan or prove a speedup. Duplicate dimension keys still multiply readings. Inspect the actual selected plan and authorized metrics after verifying uniqueness and source version. Remove unjustified permanent hints when source growth invalidates their assumptions.
from pyspark.sql import functions as F
enriched = readings.join(F.broadcast(sites), ['site_id'], 'left')Illustrative hint only. Reference size, key uniqueness, selected strategy and runtime memory must be verified independently.
Heap is not the whole process footprint
Executor memory pressure can involve JVM objects, execution structures, cached data, Python workers, native libraries and container overhead. Adding heap can reduce room for other needs if the overall allocation is unchanged. Spill is not automatically a defect: it can be part of bounded execution, though excessive spill may reveal poor sizing or distribution. Caching more data can worsen pressure. Use the deployment’s documented resource model rather than copying a heap percentage from another platform. Kubernetes, YARN and managed services expose different limits and ownership; do not assume the driver and executors share one resource budget.
A remedy must preserve the result and recovery path
Reduce unnecessary columns, fix join multiplication and choose valid pre-aggregation before requesting more capacity. For genuine key skew, prove any salting and recombination against exact small data; adaptive skew handling has operation-specific eligibility. Reconcile output after failed attempts because an external sink may have accepted partial work. Compare representative workloads after one justified change and include recovery cost, not merely one successful task. This course changes no memory, retry, shuffle or cluster settings and runs no stress test. It validates only the ratio fixture and logical remediation reasoning.
Common errors and fixes
FetchFailedException
- Why it happens
- A task cannot retrieve required shuffle data, possibly after an upstream executor or storage failure.
- How to fix it
- Inspect the first upstream failure before increasing retries.
- Prevent it next time
- Preserve stage and executor timelines with recovery evidence.
java.lang.OutOfMemoryError
- Why it happens
- The JVM cannot satisfy a required allocation within its available memory.
- How to fix it
- Identify the process, data shape and retained structures before resizing.
- Prevent it next time
- Budget driver, executor, cache and Python memory separately.
More retries repeatedly produce partial sink writes
- Why it happens
- A persistent fault and non-idempotent sink were treated as transient computation only.
- How to fix it
- Repair the cause and reconcile operation identities before replay.
- Prevent it next time
- Test failure windows in the publication protocol.
Check your understanding
1. What does max6000ms versus median200ms prove?
Show worked solution
Only that the synthetic maximum is30 times the median and the distribution has a long tail. It does not identify skew, a disk problem or garbage collection by itself. Compare per-task records, bytes, spill and placement with the result grain before selecting a remedy.
2. Can broadcasting a duplicate site dimension repair an inflated energy total?
Show worked solution
No. It changes the physical strategy while preserving the same join multiplication. Establish the authoritative reference grain first, reconcile keys and totals, and only then evaluate whether broadcasting fits the actual memory and transfer budget.
Key takeaways
- Failure classification precedes resource tuning.
- A faster or retried wrong result is still wrong.