Analytical Query Performance and Optimization Questions

This snapshot is a strong, concrete skew and spill signal, not a generic slowness. The stage does an Exchange hashpartitioning(user_id, 200) (a shuffle keyed on user_id into 200 partitions) over 1.2 billion rows, writes 1TB of shuffle data and reads back 800GB, has half its tasks spilling to disk, and shows a median task duration of 30 seconds against a maximum of 25 minutes.

Reading the numbers

• 1.2 billion rows over 200 partitions is 6 million rows per partition on average, and roughly 833 bytes of shuffle-write data per row (1TB / 1.2B), which is a plausible width for a row carrying several columns plus overhead, not itself alarming.
• The real signal is the 50x gap between the median (30s) and the maximum (25 minutes) task duration. If the data were evenly distributed across the 200 hash partitions, task durations would cluster tightly around the median. A 50x outlier on a hash-partitioned shuffle almost always means one or a few user_id values are far more frequent than the rest, so their hash partition received a disproportionate share of the 1.2 billion rows: classic data skew.
• 50% of tasks spilling to disk on top of that tells you the executor memory budget is also too tight for the partition sizes being produced, which will independently slow every task down even before skew is fixed, and will make the skewed partitions even worse because they spill the most.

Prioritized fix list

1. Confirm the skewed key(s) by running a GROUP BY user_id ORDER BY count(*) DESC LIMIT 20 (or the engine's skew-diagnosis UI) before changing anything; do not guess.
2. Mitigate the skew itself: salt the heavy key(s) so their rows spread across multiple shuffle partitions. Salting means appending a synthetic suffix, say a random integer from 0 to 7, onto the skewed key so its rows hash into several different shuffle partitions instead of piling into one; on the other side of the join, the matching row for that key is replicated once per suffix value (so the effective join key becomes (user_id, salt)) before the two sides are joined and the sub-results are combined back together. Alternatively, enable the engine's adaptive skew-join handling if available, or broadcast the smaller side of the join if one exists.
3. Fix the spill separately: increase executor memory or reduce the number of concurrent tasks per executor so each task gets a larger memory budget, since spill will still hurt even after skew is addressed if 6 million rows per partition is genuinely near the memory ceiling.
4. Re-tune shuffle partition count only after the above: 200 partitions for 1.2 billion rows may simply be too coarse regardless of skew, and increasing partition count (with the skewed key salted) reduces the amount of data any single task has to hold.

Trade-offs and pitfalls

It is tempting to jump straight to "increase shuffle partitions" because it is a one-line config change, but doing that without addressing the skewed key just spreads the non-skewed partitions thinner while the skewed one is still one giant partition (splitting 200 into 2000 buckets does nothing for a key that all hashes to the same bucket). Fix the skew first, then re-tune partition count and memory, and re-measure after each change rather than changing all three at once, since it becomes impossible to tell which change actually helped.

The danger here is a BI cache reading a partition mid-update, seeing some of the concurrent ETL jobs' writes committed and others not yet, and serving a genuinely inconsistent snapshot that mixes old and new data across the affected partitions; the fix is ensuring a cache invalidation or refresh only ever sees a fully-committed, self-consistent state, never a half-written one.

Design: atomic swap

Rather than updating the underlying table's partitions in place while queries (and the cache-refresh process) might read them mid-write, write each refresh's output to entirely new physical locations (new partition versions, or a new table snapshot), and only make the swap from the old version to the new one visible via a single atomic pointer or metadata update once every partition in that refresh window has finished writing successfully. Readers, including whatever process refreshes the BI cache, always see either the complete old state or the complete new state, never a mixture, because the atomic swap is the only thing that changes what "current" means.

Coordination mechanism

In a common cloud warehouse, this pattern is usually implemented via the table format's own snapshot/versioning mechanism (many modern table formats support atomic, all-or-nothing commits of a batch of partition changes as a single new table version) rather than something you build from scratch; if the warehouse does not expose that directly, an equivalent can be built with a staging-table-plus-atomic-rename pattern: write the refreshed partitions to a staging location, and only rename/repoint the production table reference to the staging location once every partition in the batch has written successfully, so a partial failure leaves the old, still-consistent version untouched and visible.

Tying the cache to this

The BI cache should invalidate (or refresh) based on the table's version or commit identifier changing, not on a fixed schedule or a naive "did anything in the underlying table change" signal; that way, the cache only ever picks up a fully-committed new version, and a refresh job that is still mid-write (partitions committed for some regions but not others) does not trigger a premature cache invalidation that would expose the BI layer to a half-updated state.

Trade-offs and pitfalls

This design requires the underlying storage to support cheap, atomic multi-partition commits (either natively via the table format, or built via the staging-and-rename pattern), which adds real engineering investment if the platform does not already provide it; a naive in-place partition update with no atomic swap will always be exposed to this race condition regardless of how carefully the cache-invalidation logic is written, since the inconsistency is created at the write layer, not the cache layer, no amount of cache-side cleverness fixes a write-layer race.

In Spark, a shuffle happens in two coordinated phases: on the map (write) side, each upstream task partitions its output rows by the target key (typically hashing the key modulo the number of downstream partitions), sorts them by target partition, and writes the whole task's output to a single data file plus a matching index file on local disk, with the index recording where each downstream partition's segment begins and ends inside that one file; on the reduce (fetch) side, each downstream task requests exactly the partition(s) it owns from every upstream task that produced shuffle output, over the network, and merges what it receives before continuing.

The external shuffle service

Because shuffle files are written to a task's local disk, a naive design would require the original executor process to stay alive to serve fetch requests for its shuffle output, which is fragile: if that executor is reclaimed (common under dynamic allocation, where idle executors are released to save cost) before every downstream task has fetched its data, the shuffle output becomes unavailable and the stage has to be recomputed. An external shuffle service runs independently of any single executor's lifecycle (often as a separate long-lived process per node) and serves shuffle files on the executor's behalf, so an executor can be safely released without losing shuffle data other tasks still need to fetch.

What causes many small shuffle files

Since Spark 1.2 (and exclusively since Spark 2.0, when the older hash-based shuffle manager was removed entirely), Spark's default sort-based shuffle writes exactly one data file and one index file per map task, not one file per (map task, reduce partition) pair; that per-reducer-file behavior belonged to the now-removed hash-based shuffle manager and is not how current Spark actually writes shuffle output. So the number of physical shuffle files on disk is proportional to the number of map tasks (roughly two files each), and the most direct cause of many small shuffle files today is too many small map tasks: a large number of small input splits (many tiny upstream files, or a stage inheriting an already over-partitioned output from a prior stage) produces a correspondingly large number of small map tasks, each writing its own small data and index file. Separately, on the fetch side, every reduce task still has to request its own segment from every map task's file, so the number of individual shuffle-block fetches scales with map tasks multiplied by reduce partitions even though the physical file count does not (concretely: 50 map tasks and 200 reduce partitions produce 50 x 200 = 10,000 individual shuffle-block fetches, even though there are only around 100 physical shuffle files on disk, roughly two per map task); this fetch fan-out, not raw file count, is what degrades badly when both numbers are large, a storm of many small network reads rather than literally many small files on disk.

Reducing the overhead

• Reduce the number of map tasks by combining small input splits before the shuffle stage, if the input itself is fragmented into many small files or partitions.
• Reduce the number of shuffle (reduce) partitions to a count proportional to the actual data volume rather than a fixed, overly generous default, so each map task's single shuffle file is divided into fewer, larger reducer segments and fewer, larger downstream output files result.
• Use adaptive shuffle-partition coalescing (the Adaptive Query Execution, AQE, mechanism, which observes actual post-shuffle partition sizes at runtime and merges small ones automatically, rather than relying on a static, pre-execution guess).
• Avoid triggering the shuffle at all where possible, for example replacing a shuffle join with a broadcast join when one side of the join comfortably fits in memory: eliminating a shuffle beats optimizing the file count and fetch fan-out of one that did not need to happen.

Trade-offs and pitfalls

Reducing the number of shuffle partitions to cut file-count overhead has a real ceiling: too few partitions and each one becomes too large to process efficiently (and more vulnerable to skew, since fewer partitions means each is more likely to be a meaningful fraction of the total data). The goal is balancing partition count against both file-count overhead and per-partition size, not minimizing either one in isolation, and this is exactly the kind of trade-off AQE's adaptive coalescing is designed to make automatically based on runtime statistics rather than a static guess made before the job runs.

Supporting both fast local analytics and correct global aggregates under a data-residency constraint means accepting that no single mechanism serves both well, local queries need data physically nearby to be fast, while global aggregates need data (or at least aggregated summaries of it) to be visible across regions, and residency rules constrain what can move at all.

Strategy

• Keep raw, row-level data local to its region of origin, satisfying the residency requirement directly and giving every local query full, fast access to its own region's detail without any cross-region dependency.
• Replicate only pre-aggregated, non-sensitive summaries across regions, not raw rows, for the specific metrics global dashboards need. Aggregation is often (though not always, check with legal) enough to satisfy residency rules that are concerned with individually-identifiable raw data, since a regional total or count typically does not expose the underlying records.
• Use federated queries sparingly, only for genuinely ad-hoc global analysis that cannot be anticipated and pre-aggregated in advance, understanding that each federated query pays real cross-region latency and, depending on the query engine, may itself risk pulling row-level data across the boundary if not carefully scoped to already-aggregated views.
• Materialize the global aggregates on a schedule matching the actual freshness need, rather than computing them live on every dashboard load, so the ongoing cross-region cost is the aggregation refresh, not every individual global query.

Weighing the options

Replication of pre-aggregated summaries is the right default because it keeps the residency-sensitive raw data local, keeps local query latency unaffected, and bounds cross-region cost to a controlled, scheduled refresh rather than unpredictable ad-hoc traffic. Federated queries are the right fallback specifically for the residual case, genuinely novel global questions nobody anticipated, since building a materialized aggregate for every conceivable global question in advance is not realistic. Full replication of raw data across regions is generally the wrong choice here specifically because it directly conflicts with the residency requirement; only pursue it for data explicitly cleared for that visibility.

Consistency and legal constraints

Pre-aggregated cross-region summaries are inherently eventually consistent (as fresh as their last refresh), which is usually an acceptable trade for global reporting use cases; if a global aggregate genuinely needs strong, real-time consistency, that requirement should be treated as a red flag to revisit with legal, since it likely implies a stronger cross-region data-visibility requirement than aggregation alone can satisfy, and the technical design should not quietly work around a legal question that needs an explicit answer.

Trade-offs and pitfalls

The riskiest mistake is treating "aggregated" as automatically safe for residency purposes without confirming it with whoever owns the residency requirement; a sufficiently narrow aggregate (a count of exactly one user matching rare criteria) can still be effectively identifying, so the aggregation grain itself needs review, not just the fact that aggregation happened.

Reading a slow-query profile is fundamentally the same discipline across engines, comparing what the query actually did against what it should have needed to do, but the artifact you read to get there differs by product: Snowflake's Query Profile shows an operator-by-operator breakdown with bytes scanned and pruning stats per step, BigQuery's job statistics expose bytes processed, slot-time consumed, and per-stage timing, and Redshift exposes system tables (STL_QUERY, SVL_QUERY_REPORT, STL_SCAN, SVV_TABLE_INFO) you query directly with SQL rather than a graphical profile.

The common diagnostic shape

1. Start with bytes/rows scanned versus what the filter should have needed. If a query filtering to one day scanned the whole table's history, that is a pruning failure before anything else is worth investigating.
2. Look at where time, CPU, or slot-time actually went, not just total duration. Snowflake's Query Profile breaks each operator's time into a percentage breakdown including a processing (CPU-bound) share; BigQuery's slot-time-consumed statistic is fundamentally a CPU-time proxy, since a slot is a unit of CPU and memory; Redshift's SVL_QUERY_REPORT.elapsed_time per step combined with the is_diskbased flag distinguishes a step that is CPU-bound in memory from one that spilled to disk. A query dominated by CPU-heavy computation (an expensive per-row expression, a large in-memory sort) needs a different fix than one dominated by I/O (an unpruned scan) or network (a shuffle stage).
3. Check for spill or disk-based intermediate results (Snowflake's Query Profile flags disk spillage explicitly; Redshift's SVL_QUERY_REPORT and disk-usage views show it; BigQuery surfaces this less directly but excessive shuffle bytes at a stage is the analog signal), since spill often means the query's working set does not fit the allocated compute.
4. Cross-check against table statistics/metadata freshness: an operator's estimated versus actual row counts diverging sharply points to stale statistics driving a bad plan choice.

Reading a Spark SQL plan instead

On Spark SQL specifically, there is no separate "job details" view the way BigQuery or Snowflake offer; instead you read EXPLAIN (or the Spark UI's SQL tab) to see the logical plan (what the query asks for) alongside the physical plan (how the engine intends to execute it, including Exchange nodes for shuffles and the chosen join strategy), and cross-reference that physical plan against the Spark UI's stage-level metrics (shuffle read/write, task duration) exactly as you would use Redshift's SVL_QUERY_REPORT or Snowflake's Query Profile, the artifact differs, the underlying discipline of comparing planned-versus-actual does not.

Worked example: Redshift

-- find the query (STL_QUERY has no "elapsed" or "substring" columns: compute elapsed explicitly and truncate the query text with the SUBSTRING function)
SELECT query, DATEDIFF(seconds, starttime, endtime) AS elapsed_seconds, SUBSTRING(querytxt, 1, 100) AS query_preview
FROM stl_query
WHERE userid <> 1 ORDER BY starttime DESC LIMIT 20;

-- see where its time went, per step
SELECT * FROM svl_query_report WHERE query = <query_id> ORDER BY segment, step;

-- check for a runaway scan
SELECT * FROM stl_scan WHERE query = <query_id>;

-- check table stats/skew that might explain a bad plan
SELECT * FROM svv_table_info WHERE "table" = 'fact_table';

This sequence (identify the query, see per-step time, inspect the scan step specifically, then check table-level stats) mirrors what a Snowflake Query Profile view or a BigQuery job-details page shows visually rather than via SQL.

Trade-offs and pitfalls

The single biggest mistake is reading total elapsed time and stopping there; two queries with identical wall-clock time can have completely different bottlenecks (one I/O-bound on an unpruned scan, one shuffle-bound on a skewed join), and the fix for each is unrelated to the other. Always break the total down by stage or step, and by CPU versus I/O versus network, before deciding what to change, and treat the metrics as the primary evidence over intuition about what "should" be slow.