Dataflow Spark Oom Troubleshooter

Data Flow Spark OOM Troubleshooter

Core Rule

Separate the reported failed stage from the root failure. FetchFailedException often marks the stage that discovered missing shuffle blocks; the root cause is usually the earlier executor/container death that held those blocks.

Evidence To Gather

Prefer concrete artifacts over speculation:

• Final/current adaptive physical plan and initial plan when available.
• Failed stage IDs, task metrics, shuffle read/write, input, spill, GC time, failed task host/executor.
• Executor loss reason from Spark UI, driver log, or cluster manager.
• Executor stderr/stdout around the killed executor.
• Spark configs: executor memory, memoryOverhead, executor cores, instances, shuffle partitions, AQE, broadcast thresholds, advisory partition size, off-heap.
• Spark UI SQL DAG to map stage IDs to operators.
• For the sql-endpoint/monitoring/event_logs sample, Spark History Server REST outputs such as applications, stages, executors, and environment configs.

If a decisive artifact is missing, ask for that specific item. Avoid broad log requests.

Workflow

1. Classify the failure signature:
   • exit code 137, SIGKILL, OOMKilled: container/cgroup memory exceeded.
   • java.lang.OutOfMemoryError: Java heap space: JVM heap pressure.
   • GC overhead limit exceeded: heap churn or excessive live set.
   • Direct buffer memory or Netty errors: off-heap/direct/network pressure.
   • FetchFailedException plus ExecutorDeadException: victim stage lost shuffle data from a dead executor.
2. Map the failed stage to physical operators using the Spark UI SQL DAG and plan:
   • BroadcastExchange / BroadcastHashJoin: broadcast collection and resident hashed relation.
   • HashAggregate: grouping state, distinct, spills.
   • Generate explode, especially explode(arrays_zip(...)): transient row/object expansion.
   • AQEShuffleRead coalesced: AQE may have enlarged per-task input.
   • UnsafeShuffleWriter.write: shuffle writer buffers, sorting, spill, direct/native pressure.
3. Compare data volume to concurrency:
   • Per-task pressure is roughly stage bytes divided by actual reducer/task count, then amplified by skew, row width, broadcast state, and shuffle writer overhead.
   • High executor cores multiply per-container pressure because many tasks share one JVM/container.
4. Check broadcast behavior:
   • Confirm whether repeated usage shows ReusedExchange.
   • Treat broadcast size as resident executor memory during downstream tasks, not only as a completed exchange stage.
5. Check Kubernetes/container memory:
   • Container limit is roughly spark.executor.memory + spark.executor.memoryOverhead on Kubernetes.
   • Low GC time with OOMKilled often indicates container/off-heap/native/direct/aggregate memory, not classic heap thrash.
6. Produce a root-cause statement with confidence:
   • Name the immediate failure, the upstream cause, and the plan/operator behavior that created memory pressure.
   • Distinguish mitigations that preserve broadcast from mitigations that remove it.

Analyzer Script

When the user provides plan/log files, run:

python3 scripts/analyze_spark_oom.py <file> [more-files...]

Run it from this skill directory, or use the full path to this skill's scripts/analyze_spark_oom.py. Use the script output as a triage aid, then reason from the original artifacts.

Common Recommendations

Keep recommendations tied to evidence:

• For Kubernetes OOMKilled with low GC: increase spark.executor.memoryOverhead; for large executors start near 10-15% of executor memory, higher for heavy shuffle/broadcast.
• For high shuffle read/write per task: increase spark.sql.shuffle.partitions; guard against AQE coalescing with a smaller spark.sql.adaptive.advisoryPartitionSizeInBytes or disable coalescing for the test.
• For high executor core count: prefer more executors with fewer cores, commonly 8-16 cores each, to reduce per-container concurrency and shuffle-loss blast radius.
• To preserve useful broadcast: keep broadcast enabled, but verify reuse, reduce concurrent shuffle pressure, and size overhead for resident broadcast plus shuffle/network buffers.
• For explode(arrays_zip(...)): prefer exploding/filtering the original array of structs when possible, for example explode(filter(array_col, x -> ...)), then select struct fields.
• For FetchFailed after executor death: fix the executor loss cause and consider external shuffle service, shuffle tracking, or decommissioning support if available.

For detailed signatures, read references/spark-oom-signatures.md. For the event-log monitoring sample and Spark History Server REST workflow, read references/dataflow-event-logs.md.

Source: oracle-samples/oracle-dataflow-samples — distributed by TomeVault.