Nightly Spark Join Misses Finance SLA
A retail company runs a nightly Spark batch on CDP that joins a 2.4 TB sales fact table (about 18 billion rows) to a 40 MB store dimension. The job must publish by 06:00 for finance. On three of the last five runs it finished between 07:10 and 08:20. The Spark UI shows 196 of 200 tasks finishing in 8-12 minutes, while four tasks run for 2.5-3.5 hours. The slow tasks all process the same three store_id values. The fact table is partitioned by sale_date and is not bucketed; one store_id accounts for 38% of fact rows. The cluster has 60 executors with 8 cores and 24 GB each, and the slow tasks show heavy shuffle spill. Late-arriving sales files land until 02:00, and the current job overwrites the entire sale_date partition. Analysts sometimes see duplicate rows after a rerun. The team can change Spark configuration, job code, and table layout, but cannot add cluster capacity before the next financial close.
1. What is the most likely cause of the SLA misses, and which changes would you make first? Explain trade-offs and why simply adding executors is not the first move.
Read the worked answer
The pattern is join skew, not general under-provisioning. Four tasks process three hot store_id values, and one key is 38% of the fact table. Those tasks shuffle and spill while 196 tasks finish. Adding executors gives more slots, but the hot partition remains one task, so it adds cost without fixing the critical path.
First, broadcast the 40 MB store dimension so the fact-side shuffle disappears. That size is small relative to 24 GB executor memory, but verify it fits with execution overhead. Second, enable adaptive query execution and skew join handling, and set a skew threshold that actually matches the observed partition sizes. AQE can split oversized partitions, though a 38% key may still need help. Third, for the hottest keys, salt the fact side (for example, 16 salts) and replicate the matching dimension rows, or split hot keys into a separate aggregate path. Revisit bucketing later: bucketing by store_id can help repeated joins, but it adds write cost and only helps if the bucket count matches the query pattern. Validate with stage-level metrics and a timed rerun, not just job success.
2. How would you make reruns idempotent and handle files arriving until 02:00 without putting the 06:00 deadline at more risk?
Read the worked answer
Replace whole-partition overwrite with partition-scoped, deterministic writes. Land all source files for a sale_date in a staging location, deduplicate by the business key plus the latest ingestion or sequence timestamp, and write only the affected sale_date partitions. In Hive ACID or Iceberg, use MERGE on (sale_id, sale_date) with a tie-breaker. In plain Hive, build a temp table with row_number() over (partition by sale_id order by ingest_ts desc), then insert overwrite the specific partition from the deduplicated temp table. Enable dynamic partition overwrite so a late file for one day does not rewrite every day.
Keep a control table of source file checksums and load status so a rerun skips already-committed files. Acquire a partition-level lock or use a workflow sensor so backfill and live load do not race. Set the late-data cutoff at 02:00 with a final reconciliation at 02:30; files after that go to a next-day correction path. Test by running the same date twice and comparing counts, not just checking that the job exits successfully.