Module 12: Spark & Databricks Performance Optimization

• Transformations (filter, select, join) are lazy: they build a plan
• Actions (count, write, display, collect) trigger work
• A shuffle (join, groupBy, repartition) splits a job into stages
• A stage = many tasks, one per partition, run in parallel
• Slow queries are usually: too much data read, a big shuffle, one slow (skewed) task, or too many tiny files

q = (spark.table("perf_orders")
        .filter("customer_id = 42")
        .groupBy("product_id").agg(F.sum("quantity").alias("qty")))
q.explain("formatted")

• Read the physical plan bottom to top: scan first, result last
• Words to look for:

• explain() shows the physical plan. "extended" and "formatted" show more
• SQL EXPLAIN doesn't run the query. After running, AQE may have changed the real plan

• You get these for free with built-in functions and DataFrames
• A Python UDF is a black box: no pushdown through it, extra serialization (Module 3)
• Good statistics (Delta stats, ANALYZE) make better choices
• Write the clear query. Check the plan only if it's slow

j = (spark.table("perf_orders").join(spark.table("perf_customers"), "customer_id")
          .groupBy("country").agg(F.count("*").alias("orders")))
j.collect()
j.explain()          # after running: look at isFinalPlan=true

• On by default
• Defaults: broadcast threshold 30 MB, advisory partition size 64 MB, skew factor 5 and 256 MB
• On serverless spark.sql.shuffle.partitions is auto
• You can't tune most spark.sql.adaptive.* settings on serverless: it's managed for you

• Photon = Databricks' native, vectorized engine, on by default for serverless compute and SQL warehouses
• Speeds up scans, filters, joins, aggregations, sorts, window functions and writes to Delta/Parquet
• No code change. Unsupported operations fall back to Spark. Results are the same

• Check it: in the query profile, Photon operators are purple and others grey. Execution details shows the share of time in Photon
• Fewer Python UDFs = more of the plan in Photon

(spark.table("perf_orders").withColumn("order_date", F.to_date("order_ts"))
      .write.mode("overwrite").partitionBy("order_date").saveAsTable("perf_orders_by_date"))

f = spark.table("perf_orders_by_date").filter("order_date = '2025-03-01'")
f.explain("formatted")         # look for PartitionFilters / PushedFilters in the scan
print(f.count())

• Filter on the raw column. WHERE to_date(order_ts) = ... can block pruning; WHERE order_ts >= ... AND order_ts < ... is safe
• select only needed columns before heavy steps
• Hive-style partitioning suits very large tables with a low-cardinality column. For new tables prefer liquid clustering (topic 14)

(spark.table("perf_orders").orderBy("order_ts")
      .write.mode("overwrite").saveAsTable("perf_orders_sorted"))

timed("selective filter", lambda: spark.table("perf_orders_sorted")
      .filter("order_ts BETWEEN '2025-03-01' AND '2025-03-02'").count())

• Delta stores min / max / null count per file for columns
• A filter on a column whose ranges are sorted or clustered skips most files
• Data order matters: ordered by time, a time filter skips well. Random order, no skipping

%sql
ALTER TABLE perf_orders_sorted SET TBLPROPERTIES ('delta.dataSkippingStatsColumns' = 'order_ts,customer_id')

• Long strings are truncated in stats: keep rarely filtered long text columns out
• Check files pruned in the query profile (topic 15)
• Liquid clustering is the reliable way to keep data clustered (topic 14)

from pyspark.sql.functions import broadcast

big, small = spark.table("perf_orders"), spark.table("perf_customers")

b = big.join(broadcast(small), "customer_id")       # hint: broadcast the small side
s = big.join(small.hint("merge"), "customer_id")    # force sort-merge, to compare
b.explain(); s.explain()

• Spark broadcasts automatically when a side is under the threshold (AQE: 30 MB)
• Broadcast only genuinely small tables. Too big = out of memory
• In the plan: BroadcastHashJoin with BroadcastExchange vs SortMergeJoin with two Exchange
• Join on the same type and null-safe keys. Dedupe the small side first (Module 3)

• Every Exchange writes data to disk and moves it across the network
• Shuffle triggers: join, groupBy, distinct, orderBy, repartition, window functions

# less shuffled data: filter and aggregate before the join
daily = (spark.table("perf_orders").filter("quantity > 1")
              .groupBy("customer_id").agg(F.sum("quantity").alias("qty")))
result = daily.join(spark.table("perf_customers"), "customer_id")

• Too many shuffle partitions = tiny tasks; too few = huge tasks and spill. auto handles this
• orderBy only at the end, only when output order matters

sk = spark.table("perf_skewed")
sk.groupBy("customer_id").count().orderBy(F.desc("count")).show(3)    # key 1 has ~70% of rows

Salting a skewed aggregation: two steps.

N = 8
salted = (sk.withColumn("salt", F.pmod(F.hash("order_id"), F.lit(N)))
            .groupBy("customer_id", "salt").agg(F.count("*").alias("n"))     # partial, spread out
            .groupBy("customer_id").agg(F.sum("n").alias("n")))               # final, small

Salting a join (when AQE doesn't fix it): replicate the small side across salt values.

small = spark.table("perf_customers")
small_salted = small.crossJoin(spark.range(N).select(F.col("id").cast("int").alias("salt")))
big_salted = sk.withColumn("salt", F.pmod(F.hash("order_id"), F.lit(N)))
joined = big_salted.join(small_salted, ["customer_id", "salt"])

• Signs of skew: one task far slower than the median, one long bar in the profile
• Try first: AQE skew handling (on), a filter that removes the hot key, a broadcast join
• Salting costs complexity: use it when the others fail

df = spark.table("perf_orders")
wide = df.repartition(16, "customer_id")      # full shuffle, even sizes, hash by key
narrow = df.coalesce(2)                       # merge partitions, no shuffle

• repartition("key") groups equal keys: useful before a write that is read by key
• Don't coalesce(1) big data: all work moves to one task
• Don't repartition "just in case". Measure first
• Serverless manages partitions: getNumPartitions() and the RDD API aren't available

• df.cache() / persist() raise an error on serverless
• Delta tables get automatic disk caching of data on classic compute and SQL warehouses
• SQL warehouses also keep a query result cache (an identical query returns instantly)

stage = (spark.table("perf_orders").join(spark.table("perf_customers"), "customer_id")
              .groupBy("country", "product_id").agg(F.sum("quantity").alias("qty")))
stage.write.mode("overwrite").saveAsTable("perf_stage")        # the 'cache'

by_country = spark.table("perf_stage").groupBy("country").agg(F.sum("qty").alias("qty"))
by_product = spark.table("perf_stage").groupBy("product_id").agg(F.sum("qty").alias("qty"))

• A written table is also inspectable, restartable and shareable
• On classic compute, cache only data used several times, and unpersist() when done

(spark.table("perf_orders").repartition(100)
      .write.mode("overwrite").saveAsTable("perf_small_files"))

display(spark.sql("DESCRIBE DETAIL perf_small_files").select("numFiles", "sizeInBytes"))

%sql
OPTIMIZE perf_small_files

display(spark.sql("DESCRIBE DETAIL perf_small_files").select("numFiles", "sizeInBytes"))   # far fewer files

• Many tiny files = slow listing and many small tasks
• Causes: too many partitions, frequent small appends, streaming micro-batches, over-partitioned tables
• Fixes, in order of preference:

• Never leave coalesce(1) or hundreds of partitions on regular writes

%sql
CREATE OR REPLACE TABLE perf_orders_lc CLUSTER BY (customer_id, order_ts) AS
SELECT * FROM perf_orders

%sql
SELECT count(*) FROM perf_orders_lc WHERE customer_id = 42     -- compare files pruned in the profile

• Up to 4 clustering keys: choose columns used in filters
• Data is clustered on OPTIMIZE. Predictive optimization does it for managed tables
• CLUSTER BY AUTO lets Databricks choose keys (needs predictive optimization)
• Not combined with partitioning or ZORDER

Open it: under a cell, See performance → pick a statement → See query profile. Or Query History → the query.

• Red flags: scan with little pruning, a join producing far more rows than its inputs, a shuffle with one huge partition, long scheduling time
• Your own timing with timed() is a quick check; the profile says why
• Queries served from the result cache have no profile

• Classic: executor memory splits into execution (shuffles, joins) and storage (cache)
• Spill (data written to disk during a shuffle or sort) means memory is short: more memory per core, fewer partitions per task, or less data
• Out-of-memory causes: collecting to the driver, a huge broadcast, a skewed partition, very wide rows
• Serverless has few settable configs: spark.sql.shuffle.partitions, spark.sql.session.timeZone, spark.sql.ansi.enabled, spark.sql.files.maxPartitionBytes, timeouts
• If a serverless query runs out of memory: reduce the data, fix skew, avoid collect(), or use a larger/high-memory option where offered

• Many serverless limits exist (Module 2): no RDDs, no caching, few Spark configs
• Free Edition has a daily compute quota: a wasteful query uses it up. Use LIMIT while exploring