Performance: Optimization & Shuffles

Catalyst Optimizer & Predicate Pushdown

Because DataFrames carry a known schema (unlike opaque RDD lambda functions), Catalyst can analyze the whole logical plan and apply optimizations — predicate pushdown, column pruning, join reordering — automatically. Predicate pushdown moves filter conditions as close to the data source as possible; against a columnar Parquet file, this lets whole chunks be skipped during the READ itself using per-row-group statistics, rather than reading everything and filtering afterward.

Shuffles & Data Skew

A shuffle redistributes data across partitions/nodes over the network — required by groupBy, join, and distinct when related data isn't already co-located. It's often the most expensive part of a job, especially with data skew: if a few keys have disproportionately many rows, their partitions become huge relative to others, and the whole job waits on that one overloaded task. Salting a skewed join key (adding a random suffix to spread a hot key across synthetic sub-keys) is a common fix.

Broadcast Joins

from pyspark.sql.functions import broadcast

# When one side is small enough to fit in memory, broadcast it to every
# executor — the large side's partitions join LOCALLY, no shuffle needed
# for the large DataFrame at all.
orders.join(broadcast(small_lookup_table), 'category_id')

# Spark auto-selects broadcast joins below a size threshold:
# spark.conf.set('spark.sql.autoBroadcastJoinThreshold', 10 * 1024 * 1024)

Caching, Partitioning & UDFs

# cache()/persist() is ALSO lazy — nothing is stored until the first
# action after it runs; calling cache() alone does no work by itself
df_cached = df.filter(col('active') == True).cache()
df_cached.count()   # this action materializes and stores the cache
df_cached.show()    # this reuses the cached data, no recomputation

# repartition() does a full shuffle — can increase OR decrease partitions
df.repartition(200, 'region')

# coalesce() avoids a full shuffle — only merges partitions, decrease-only,
# cheap way to avoid thousands of tiny output files before a write
df.coalesce(10).write.parquet('output/')

# UDFs bypass Catalyst's optimizations — prefer a built-in function first
from pyspark.sql.functions import udf
from pyspark.sql.types import StringType

@udf(returnType=StringType())
def normalize(text):
    return text.strip().lower()