Performance Tuning — Spark, SQL, and Pipeline Optimisation

APPLICATION → one SparkContext (or SparkSession)
  JOB → one per action (collect(), count(), write(), show())
    STAGE → one per shuffle boundary
      TASK → one per partition (runs on one executor core)

ONE ACTION = ONE JOB:
  df.write.parquet('/path')    ← triggers one job
  df.count()                   ← triggers another job (separate action)
  df.cache()                   ← does NOT trigger a job — lazy evaluation!
  df.cache().count()           ← triggers a job that materialises + counts

SHUFFLE (= new stage boundary):
  groupBy() + agg()    ← rows with same key must go to same partition
  join()               ← rows with same join key must meet on same node
  distinct()           ← duplicates across partitions must compare
  repartition(n)       ← explicit redistribution
  orderBy()            ← global sort requires all data to sort together

NO SHUFFLE (= same stage):
  filter()             ← each partition filtered independently
  select()             ← each partition projected independently
  withColumn()         ← row-level computation per partition
  map() / flatMap()    ← element-level operations
  limit()              ← takes N rows (but beware: final sort may shuffle)

EXAMPLE EXECUTION PLAN:
  df.filter(col('date') == '2026-03-17')   ← Stage 1: filter (no shuffle)
    .join(dim, on='store_id', how='left')  ← Stage 2: join (shuffle!)
    .groupBy('city')                       ← Stage 3: aggregate (shuffle!)
    .agg(sum('revenue'))
    .write.parquet('/gold/daily')          ← triggers all stages
  Spark creates 3 stages. Stage 2 and 3 each wait for the previous
  stage's shuffle to complete.

PARTITIONS — the unit of parallelism:
  One task processes one partition. More partitions = more parallelism
  (up to available cores). Too few: executor cores idle. Too many: shuffle
  and task-scheduling overhead.

  RECOMMENDED PARTITION SIZE: 100-200 MB after reading/filtering
  Total cores in cluster × 2-4 = good default partition count

  Default shuffle partitions: spark.sql.shuffle.partitions = 200
  200 is too low for large datasets, too high for small ones. Tune per job:
    spark.conf.set('spark.sql.shuffle.partitions', '400')

ADAPTIVE QUERY EXECUTION (AQE — Spark 3.0+):
  spark.conf.set('spark.sql.adaptive.enabled', 'true')
  AQE automatically adjusts partition count after each shuffle based on
  actual data sizes. Reduces need for manual tuning. ALWAYS enable in production.

Each row = one stage. Key columns:
  Duration:      total wall-clock time for this stage
  Input:         bytes read from storage (I/O bound if very high)
  Shuffle Read:  bytes read from previous stage's shuffle (network bound)
  Shuffle Write: bytes written to next stage's shuffle (network bound)
  Spill (Mem/Disk): data that didn't fit in memory, written to disk

RED FLAGS:
  Stage takes 30 min, Input = 2 TB → I/O bound, need better partitioning
  Stage has Spill = 50 GB → memory bound, increase executor memory
  Stage has Shuffle Read = 500 GB → network bound, consider broadcast

TASKS TAB (inside a stage):
  Duration histogram: should be relatively uniform across tasks.
  ONE TASK IS 10× SLOWER THAN OTHERS → data skew (key imbalance)

EXECUTORS TAB:
  Cores used: should be near max during active stages
  Memory used / total: if consistently > 80% → consider more memory
  Task time vs GC time: if GC > 10% of task time → memory pressure

== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- == Current Plan ==
   HashAggregate(keys=[city], functions=[sum(revenue)])
   +- Exchange hashpartitioning(city, 200)    ← SHUFFLE HERE (Stage boundary)
      +- HashAggregate(keys=[city], functions=[partial_sum(revenue)])
         +- BroadcastHashJoin [store_id], [store_id], LeftOuter, ...
            :- Filter (date = 2026-03-17)                   ← no shuffle
            :  +- FileScan parquet (orders) PushedFilters=[date=2026-03-17]
            +- BroadcastExchange HashedRelationBroadcastMode  ← broadcast dim
               +- FileScan parquet (dim_store)

Reading this plan: FileScan reads orders (filter pushed to the file reader).
BroadcastExchange broadcasts dim_store (small) to all executors.
BroadcastHashJoin: join without shuffle — fast.
Exchange before HashAggregate: one shuffle, for city-level aggregation.
Total: 2 stages, 1 shuffle, 1 broadcast. Clean plan.

df.write \
  .partitionBy('order_date', 'store_id') \
  .parquet('s3://freshcart-lake/silver/orders/')

# Creates:
#   silver/orders/order_date=2026-03-17/store_id=ST001/part-00001.parquet
#   silver/orders/order_date=2026-03-17/store_id=ST002/part-00001.parquet
#   silver/orders/order_date=2026-03-16/store_id=ST001/part-00001.parquet

# BENEFIT — partition pruning at read time:
spark.read.parquet('s3://...') \
    .filter(col('order_date') == '2026-03-17') \
    .filter(col('store_id')   == 'ST001')
# → Spark reads ONLY .../order_date=2026-03-17/store_id=ST001/
# → 99% less I/O if data has many dates and stores

CHOOSING PARTITION COLUMNS:
  ✓ Columns most commonly used in WHERE filters
  ✓ Low-to-medium cardinality (date: 365 values/year — good)
  ✗ High cardinality (customer_id: millions — too many small files)
  ✓ Columns whose values are known at write time (not derived)

FILE SIZE WITHIN PARTITIONS:
  Target: 100-500 MB per file (before compression)
  Too small: millions of tiny files → S3 LIST API overhead → slow reads
  Too large: low parallelism → fewer tasks → underutilised cluster
  Use OPTIMIZE (Delta Lake) to compact small files into target size:
    OPTIMIZE delta.`s3://freshcart/silver/orders`
        WHERE order_date >= '2026-03-01';

# Read partitioned data — Spark creates one task per file:
df = spark.read.parquet('s3://freshcart/silver/orders/')
df.rdd.getNumPartitions()   # might be 2,000 (one per file)

# Too many small partitions → too much overhead:
df = df.coalesce(200)       # reduce without shuffle (downstream only)

# Repartition by join key — align for co-located joins:
df = df.repartition(400, col('store_id'))
dim = dim.repartition(400, col('store_id'))
result = df.join(dim, on='store_id', how='left')
# Spark detects both DataFrames are partitioned by store_id
# → uses SortMergeJoin without re-shuffling either side

spark.conf.set('spark.sql.adaptive.skewJoin.enabled', 'true')
spark.conf.set('spark.sql.adaptive.skewJoin.skewedPartitionFactor', '5')
# AQE automatically splits skewed partitions and handles the skewed key.

# DIAGNOSIS: Spark UI → Stages → Tasks → duration histogram
# One task 10× longer than others → skew on the groupBy/join key

from pyspark.sql import functions as F
SALT_FACTOR = 10   # split skewed key into 10 sub-partitions

# Left side: add random salt 0-9 to each row
df_salted = df.withColumn(
    'store_id_salted',
    F.concat(col('store_id'), F.lit('_'),
             (F.rand() * SALT_FACTOR).cast('int').cast('string'))
)
# Right side: explode into 10 copies with each salt value
dim_exploded = dim.crossJoin(
    spark.range(SALT_FACTOR).select(F.col('id').cast('string').alias('salt'))
).withColumn(
    'store_id_salted',
    F.concat(col('store_id'), F.lit('_'), col('salt'))
)
result = df_salted.join(dim_exploded, on='store_id_salted', how='left')
# Each of the 10 salted ST001 sub-partitions joins independently

Used when: one table fits in executor memory
Threshold: spark.sql.autoBroadcastJoinThreshold = 10 MB (default)
Mechanism: small table broadcast to ALL executors → hash table in memory
           large table stays in place → each partition queries the hash table
No shuffle needed → fastest join type. Limitation: small table must fit
in memory × number of executors.

WHEN TO USE:
  fact_orders (500M rows) JOIN dim_store (10 stores) → BROADCAST dim_store
  fact_orders (500M rows) JOIN dim_date (11,000 rows) → BROADCAST dim_date

FORCING BROADCAST (when Spark doesn't auto-detect):
  from pyspark.sql.functions import broadcast
  result = df_orders.join(broadcast(df_dim_store), on='store_id', how='left')

TUNING THRESHOLD:
  spark.conf.set('spark.sql.autoBroadcastJoinThreshold', str(100 * 1024 * 1024))
  # 100 MB — broadcast tables up to 100 MB automatically

Used when: both tables are large, cannot broadcast either
Mechanism: (1) shuffle both DataFrames by join key to same partitions,
           (2) sort both sides within each partition, (3) merge-join.
Cost: 2 shuffles + 2 sorts → most expensive join type.
Benefit: handles arbitrarily large tables.

OPTIMISATION: pre-sort both sides on the join key before the join
  df_orders = df_orders.repartition(400, col('store_id')) \
                       .sortWithinPartitions('store_id')
  df_events = df_events.repartition(400, col('store_id')) \
                       .sortWithinPartitions('store_id')
  result = df_orders.join(df_events, on='store_id', how='inner')
  # Spark can use SortMergeJoin without re-shuffling either side

Used when: one table is smaller but not small enough to broadcast
Mechanism: shuffle both sides, build hash table from smaller side,
           probe hash table with larger side rows.
Better than SMJ when: build side is significantly smaller than probe side.
Worse than BHJ: still requires a shuffle.

FORCING SHJ:
  result = df_orders.join(
      df_medium.hint('shuffle_hash'), on='store_id', how='left'
  )

A Cartesian product (CROSS JOIN or missing join condition) multiplies rows.
10,000 orders × 10,000 products = 100,000,000 rows.
10M orders × 10K products = 100,000,000,000 rows → OOM / never finishes.

SPARK PROTECTION:
  spark.conf.set('spark.sql.crossJoin.enabled', 'false')  # default: raises error

WHEN CARTESIAN IS INTENTIONAL (and safe):
  df.crossJoin(spark.range(10))  # explode each row 10× for salting
  Small × small (e.g., 12 months × 10 stores = 120 rows) is fine.

BAD: join 500M orders to 10M payments, then filter to one day
  df.join(payments, on='order_id').filter(col('date') == '2026-03-17')

GOOD: filter orders to one day (500K rows) THEN join to payments
  df.filter(col('date') == '2026-03-17').join(payments, on='order_id')
  # 500K rows join to payments instead of 500M rows → 1000× less shuffle data

-- SLOW: function on date column prevents micro-partition pruning
SELECT * FROM silver.orders
WHERE DATE_TRUNC('day', created_at) = '2026-03-17';
-- Snowflake cannot compare the function result to partition min/max.
-- Result: full table scan. 10,000 micro-partitions → 10,000 scanned.

-- FAST: range filter on raw column enables pruning
SELECT * FROM silver.orders
WHERE created_at >= '2026-03-17'::TIMESTAMPTZ
  AND created_at <  '2026-03-18'::TIMESTAMPTZ;
-- Result: 14 micro-partitions scanned out of 10,000. 99.9% pruning.

-- SAME PROBLEM IN BIGQUERY:
-- SLOW: WHERE DATE(created_at) = '2026-03-17'
-- FAST: WHERE created_at >= '2026-03-17' AND created_at < '2026-03-18'

-- SLOW: reads all 200 columns
SELECT * FROM fct_orders_wide WHERE date = '2026-03-17';
-- BigQuery bills for ALL columns × ALL rows. Snowflake reads all
-- column micro-partition data.

-- FAST: only read needed columns
SELECT order_id, store_id, order_amount, customer_tier
FROM fct_orders_wide
WHERE date = '2026-03-17';
-- ~200× less I/O for a 200-column table.

-- SLOW for large datasets — DISTINCT sorts/hashes all values:
SELECT DISTINCT customer_id FROM silver.orders WHERE date = '2026-03-17';

-- FASTER for counting:
SELECT COUNT(DISTINCT customer_id) FROM silver.orders WHERE date = '2026-03-17';

-- FASTEST — HyperLogLog approximation (fine for most dashboards):
SELECT APPROX_COUNT_DISTINCT(customer_id) FROM silver.orders WHERE date = '2026-03-17';
-- ~2% error, 100× faster for large datasets.

-- SLOW: correlated subquery runs once per order row
SELECT o.order_id, o.order_amount,
    (SELECT AVG(order_amount) FROM silver.orders o2
     WHERE o2.store_id = o.store_id AND o2.date = o.date)
     AS store_daily_avg
FROM silver.orders o;
-- For 500K orders: runs the subquery 500K times. Extremely slow.

-- FAST: window function, computed once over all rows
SELECT order_id, order_amount,
    AVG(order_amount) OVER (PARTITION BY store_id, date) AS store_daily_avg
FROM silver.orders;
-- Window function scans data once. 1000× faster.

-- SLOW: two full scans
SELECT 'delivered' AS status, COUNT(*) FROM orders WHERE status = 'delivered'
UNION ALL
SELECT 'cancelled' AS status, COUNT(*) FROM orders WHERE status = 'cancelled';

-- FAST: conditional aggregation, one scan
SELECT
    COUNT(CASE WHEN status = 'delivered' THEN 1 END) AS delivered_count,
    COUNT(CASE WHEN status = 'cancelled' THEN 1 END) AS cancelled_count
FROM silver.orders;

-- SLOW: subquery to filter window function result
SELECT order_id, order_amount, row_num FROM (
    SELECT *, ROW_NUMBER() OVER (PARTITION BY store_id ORDER BY order_amount DESC)
              AS row_num
    FROM silver.orders
) WHERE row_num = 1;

-- FAST: QUALIFY (Snowflake-native — eliminates the subquery)
SELECT order_id, order_amount
FROM silver.orders
QUALIFY ROW_NUMBER() OVER (PARTITION BY store_id ORDER BY order_amount DESC) = 1;

{{ config(
    materialized='incremental',
    incremental_strategy='append',
    unique_key='order_id',
) }}
SELECT * FROM {{ ref('stg_orders') }}
{% if is_incremental() %}
    WHERE ingested_at > (SELECT MAX(ingested_at) FROM {{ this }})
{% endif %}

USE WHEN: fact tables where rows are never updated — event logs,
          append-only CDC events, immutable audit records.
AVOID WHEN: rows can be updated (orders change status) → creates duplicates.

{{ config(
    materialized='incremental',
    incremental_strategy='merge',
    unique_key='order_id',
    merge_update_columns=['status', 'updated_at', 'delivered_at'],
) }}
SELECT * FROM {{ ref('stg_orders') }}
{% if is_incremental() %}
    WHERE updated_at > (
        SELECT MAX(silver_updated_at) - INTERVAL '30 minutes' FROM {{ this }}
    )
{% endif %}

USE WHEN: rows can change over time (status changes, updated attributes).
merge_update_columns limits how many columns are updated per match —
without it, all columns are updated even when unchanged, which is wasteful.
The 30-minute overlap window catches late-arriving Bronze rows.

{{ config(
    materialized='incremental',
    incremental_strategy='insert_overwrite',
    partition_by={'field': 'order_date', 'data_type': 'date', 'granularity': 'day'},
) }}
SELECT * FROM {{ ref('stg_orders') }}
{% if is_incremental() %}
    WHERE order_date >= CURRENT_DATE - 2  -- rebuild last 2 days
{% endif %}

USE WHEN: large time-partitioned tables where partition-level replacement
          is more efficient than row-level merge.
BEST FOR: BigQuery (native partition-level overwrite, very cheap). Also
          effective on Spark Delta Lake (replaces whole partition files).
AVOID WHEN: multiple keys updated across many partitions → merge is better.

{{ config(
    materialized='incremental',
    incremental_strategy='delete+insert',
    unique_key='order_id',
) }}
-- dbt generates:
-- DELETE FROM {{ this }} WHERE order_id IN (SELECT order_id FROM __new_rows)
-- INSERT INTO {{ this }} SELECT * FROM __new_rows
USE WHEN: merge is not supported by the target database adapter.

CHOOSING THE RIGHT STRATEGY:
  Event log (never updates):           append
  Entity current state (updates):      merge
  Large time-series, few key changes:  insert_overwrite by date partition
  Non-merge-supporting DB:             delete+insert

INCREMENTAL FILTER WINDOW: must be wide enough to catch late-arriving rows.
A 30-minute overlap ensures rows arriving slightly after the last run are
still processed. For sources with up to 24h late arrival: use 25h overlap.

A dbt incremental merge writes a few thousand rows per run. Each run
appends small Parquet files to the Delta table. After 90 days of daily
runs: 90 small files in the partition, each requiring a separate S3 GET.
  Reading 100 columns from 90 × 5 MB files  = 9,000 S3 GET requests
  Reading 100 columns from 1 × 450 MB file  =   100 S3 GET requests
→ 90× more S3 API calls → much slower reads. After a year of hourly runs
on a busy table: 8,760 files — S3 LIST alone takes seconds before reading starts.

DIAGNOSIS (Delta Lake):
  DESCRIBE HISTORY silver.orders;
  -- Look at numFiles per version — rapidly growing count = small file problem
  SELECT file_path, size FROM silver.orders.files ORDER BY size ASC LIMIT 20;
  -- Many files under 1 MB = small file problem

-- Compact all small files in a partition into target size (256 MB default):
OPTIMIZE silver.orders WHERE order_date = '2026-03-17';

-- Z-ORDER combines compaction with co-location by column:
OPTIMIZE silver.orders ZORDER BY (store_id, order_date);
-- Files with similar store_id and order_date values are co-located.
-- Queries filtering by store_id skip ~90% of files after Z-ORDER.

-- AUTOMATING IN AIRFLOW — run after the daily dbt transformation:
optimize_silver = BashOperator(
    task_id='optimize_silver_orders',
    bash_command='databricks jobs run-now --job-id optimize_silver_orders_job',
)
dbt_silver >> dbt_gold >> optimize_silver

-- VACUUM: remove files no longer referenced by Delta:
VACUUM silver.orders RETAIN 168 HOURS;  -- keep 7 days for time travel

# BAD: silver.orders scanned TWICE in the same pipeline run
silver_orders = spark.read.format('delta').load('/silver/orders')
revenue_df  = silver_orders.filter(...).groupBy('store').agg(sum('amount'))
customer_df = silver_orders.filter(...).groupBy('customer').agg(count('*'))
# Spark reads /silver/orders twice from S3 — 2× the I/O.

# GOOD: cache after the first read, use for both downstream operations
silver_orders = spark.read.format('delta').load('/silver/orders')
silver_orders.cache()
silver_orders.count()   # trigger materialisation (eagerly cache)

revenue_df  = silver_orders.filter(...).groupBy('store').agg(sum('amount'))
customer_df = silver_orders.filter(...).groupBy('customer').agg(count('*'))
silver_orders.unpersist()  # release memory after use — important!

WHEN TO CACHE:
  ✓ Same DataFrame used 2+ times downstream in the same pipeline run
  ✓ Expensive intermediate result (join result) reused
  ✗ DataFrame only used once — cache adds overhead without benefit
  ✗ Very large DataFrames that don't fit in memory — spills to disk, slower

# GOOD: filter on the partition column directly at read time
df = spark.read.format('delta').load('/silver/orders') \
     .filter(col('order_date') == '2026-03-17')
# Spark reads ONLY the order_date=2026-03-17 partition directory —
# this partition-pruning happens automatically for column filters that
# match the partitionBy() columns used at write time.

# For non-partition column filters on Parquet:
spark.conf.set('spark.sql.parquet.filterPushdown', 'true')  # default: true
# Pushes row-group level filters into the Parquet reader.

# Memory-intensive workloads (large joins, wide aggregations):
executor_memory     = '16g'   # 16 GB per executor
executor_cores      = 4       # 4-5 cores per executor is the rule of thumb
overhead_memory     = '2g'    # ~10-15% of executor_memory

spark = SparkSession.builder \
    .config('spark.executor.memory',         '16g') \
    .config('spark.executor.cores',          '4') \
    .config('spark.executor.memoryOverhead', '2g') \
    .config('spark.driver.memory',           '8g') \
    .config('spark.sql.adaptive.enabled',    'true') \
    .config('spark.sql.adaptive.coalescePartitions.enabled', 'true') \
    .config('spark.sql.shuffle.partitions',  '400') \
    .getOrCreate()

# repartition(n): full shuffle, creates exactly n equal partitions.
#   Use when data is severely unbalanced or you need a specific count.
# coalesce(n): no shuffle, merges existing partitions.
#   Use when reducing partition count AFTER filtering — avoids network traffic.

df = spark.read.parquet(...)                       # 2,000 partitions
    .filter(col('date') == '2026-03-17')            # 95% of partitions now empty
df = df.coalesce(50)   # merge 2,000 into 50 without shuffle

WHEN TO REPARTITION: before a join (co-partitioning both sides on the join
  key), before orderBy, or when partition sizes are very uneven.
WHEN TO COALESCE: after an aggressive filter, or before writing to reduce
  file count. Never coalesce BEFORE a shuffle operation — it's wasted.

The Silver orders pipeline runs from 06:00 ET and is supposed to complete by 07:30 ET, giving Gold 30 minutes before analysts arrive. It has been completing at around 10:00 ET. The data team is asked to fix it. The pipeline processes 180 million orders in Bronze, transforming them to Silver via a Spark job on a 10-node cluster.

Step 1-2 — finding the skewed key

Stage 1 (file read + filter): 3 min    ← reasonable
Stage 2 (join with dim_store): 2.5 hr  ← THE BOTTLENECK
Stage 3 (aggregation):         35 min

Stage 2 Tasks: 1 task = 142 min, all others = 8-12 min.
ONE TASK IS 18× SLOWER → classic data skew.

# Check the join key distribution:
df.groupBy('store_id').count().orderBy('count', ascending=False).show(10)

ST001  148,000,000  ← ONE store has 148M of 180M rows (82%)!
ST002    4,200,000
ST003    3,800,000
... (remaining 9 stores share 28M rows)

ST001 is FreshCart HQ — all online orders route through this store_id.
The join on store_id puts all 148M ST001 rows in one partition.

Step 3 — fixing skew with AQE

spark.conf.set('spark.sql.adaptive.enabled', 'true')
spark.conf.set('spark.sql.adaptive.skewJoin.enabled', 'true')
spark.conf.set('spark.sql.adaptive.skewJoin.skewedPartitionFactor', '3')
spark.conf.set('spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes',
               str(256 * 1024 * 1024))  # flag partitions > 256 MB as skewed
# AQE splits the skewed ST001 partition into multiple sub-partitions.

RE-RUN RESULT: Stage 2 = 38 min (was 2.5 hr). 4× better. Still not enough.

Step 4 — fixing shuffle partition count

Stage 3: all 200 tasks, each taking 10-15 min. Input per task ~80 MB
(reasonable), but shuffle.partitions = 200 (default) for 180M rows
= 900K rows per partition — not enough parallelism.

FIX: increase shuffle partitions
spark.conf.set('spark.sql.shuffle.partitions', '800')
# 800 partitions for 180M rows = 225K rows per partition — 4× more parallelism

RE-RUN RESULT: Stage 3 = 9 min (was 35 min). Stage 2 = 34 min.

Step 5 — forcing the broadcast join

# After AQE: no more extreme skew, but 34 min for a join with dim_store?
# dim_store has 10 rows — it should be broadcast!
spark.conf.get('spark.sql.autoBroadcastJoinThreshold')  # = '10485760' (10 MB)

# dim_store is loaded from a Delta table with no updated table statistics —
# Spark estimates dim_store = 500 MB (wrong), so broadcast never triggers.

# FIX: force the broadcast hint
dim_store_df = spark.read.format('delta').load('/silver/dim_store')
orders_with_store = df_orders.join(
    broadcast(dim_store_df), on='store_id', how='left'
)

RE-RUN RESULT: Stage 2 = 6 min (was 34 min after AQE alone).

Step 6 — eliminating a redundant read, final results

bronze_orders = spark.read.format('delta') \
    .load('/bronze/orders') \
    .filter(col('_bronze_date') == run_date)
bronze_orders.cache()
bronze_orders.count()   # materialise once, reuse for both downstream models

FINAL PIPELINE TIMES:
Stage 1 (read + filter):  3 min
Stage 2 (join):           6 min  (was 2.5 hours)
Stage 3 (aggregate):      9 min  (was 35 min)
Stage 4 (second model):   4 min  (cache hit — was 12 min)
Total: 22 min (was 4 hours) — 11× faster. SLA now completes at 06:22 ET.

SUMMARY OF FIXES APPLIED:
1. AQE skew join:          2.5 hr → 38 min  (data skew resolved)
2. Broadcast dim_store:    38 min → 6 min   (wrong join strategy)
3. Shuffle partitions 800: 35 min → 9 min   (too few partitions)
4. Cache Bronze read:      12 min → 4 min   (redundant S3 read eliminated)