Spark SQL Join Misestimated and Skewed on NULL store_id

Spark chooses a broadcast hash join when the estimated size of one side is below spark.sql.autoBroadcastJoinThreshold, default 10 MB. Without metastore statistics, the optimizer falls back to a default size estimate, and the Hive relation here is estimated at 6 GB. That is far above the threshold, so store_dim is joined with a sort-merge join, which means the 4 TB fact side and the dimension both go through a shuffle. Shuffling a 2 MB table across the cluster is pure waste and is the direct cause of most of the 1.2 TB spill.

Immediate fix: collect statistics. Run ANALYZE TABLE store_dim COMPUTE STATISTICS FOR COLUMNS and the same for customer_dim and the fact table (or at least the join and filter columns). With correct stats, store_dim at 2 MB is auto-broadcast and the join disappears from the shuffle. As a belt-and-braces measure, add the /*+ BROADCAST(s) */ hint for this query so it is stable even if stats drift.

The 400 MB dimension must be treated differently. It is above the 10 MB threshold, and broadcasting 400 MB to every executor is dangerous: with 8 task slots per executor, a broadcast hash table can be shared per executor, but the build side plus execution memory on a 32 GB node can trigger the same GC failures. Keep the sort-merge join for customer_dim, but let AQE handle skew and partition sizing. If measurements show the compressed 400 MB table is only tens of MB in memory and executors have headroom, raising the threshold to 64 MB and broadcasting it is a legitimate follow-up experiment, validated on a subset of partitions.

All NULL keys hash to the same partition, so one task receives 30% of the month's rows while its peers get a fraction of that; that is the 90-minute task. AQE's skew join handling can split oversized partitions, but a single NULL group cannot be split by key. The robust fix is to keep NULLs out of the join: aggregate the NULL rows separately (they collapse to one 'unknown store' group), then union the result with the main join, which now runs only on non-NULL keys. An alternative is to replace NULL with a sentinel and salt it with a random suffix, aggregate in two stages, and combine. The separate-aggregation route is cheaper and clearer.

Sizing arithmetic: the table is 4 TB over 1,095 daily partitions, roughly 3.7 GB per partition, so a one-month filter reads about 110 GB. After the projection and partial aggregation, shuffle output is commonly 1-2x the scanned bytes; assume about 150-200 GB. Targeting 128-200 MB per shuffle partition gives 150 GB / 128 MB ~ 1,200 partitions, so set spark.sql.shuffle.partitions=1200 (or 2,000 and let AQE coalesce downward). The current 200 produces 750 MB-1 GB partitions before skew, which explains the spill.

Settings: spark.sql.adaptive.enabled=true, spark.sql.adaptive.coalescePartitions.enabled=true, spark.sql.adaptive.skewJoin.enabled=true, and keep spark.sql.adaptive.advisoryPartitionSizeInBytes near 128 MB. Executor sizing: on 32 GB nodes with 8 cores, use 2 executors of ~12-14 GB with spark.memory.fraction=0.6, leaving about 7-8 GB for execution and shuffle. Raising executor memory alone will not fix GC overhead if the partition count and join strategy are wrong; the GC error during shuffle fetch is a symptom of oversized partitions plus a large build side.

Each Spark task writes one file per dynamic partition it touches. With 200 shuffle partitions (or 1,200 after resizing) writing into about 30 daily partitions, every task that holds rows for a given dt opens its own file for that dt, so a partition of roughly 3.7 GB ends up as dozens to hundreds of 3 MB files. Across the month that is tens of thousands of files, which stresses the NameNode, slows listing and planning, and makes every subsequent read schedule far more tasks than needed.

Write-side changes:

1. Repartition by the partition column before writing: DISTRIBUTE BY dt (or REPARTITION(dt)) so each output partition is produced by a small, controlled number of tasks rather than by all of them. Keep the shuffle partition count sensible so you do not create one giant task per dt.
2. Cap file size where needed with spark.sql.files.maxRecordsPerFile, and use spark.sql.files.openCostInBytes to make the writer account for file-open cost when packing rows.
3. Consider INSERT OVERWRITE of the affected partitions rather than append, so reruns do not multiply files.

Ongoing maintenance: run a nightly or weekly compaction job that reads the small-file partitions and rewrites them at 128-256 MB, deleting the originals after verification. Then keep statistics current: ANALYZE TABLE store_sales PARTITION(dt='...') COMPUTE STATISTICS for new partitions, so the same misestimation that caused the join problem does not reappear at the partition level. Finally, monitor file count per partition and average file size as service-level indicators; a rising file count with flat data volume is the early warning that the writer has regressed.