Joins and shuffles are where big-data jobs get slow or fall over. Understand what a shuffle costs, join efficiently, and write output that stays fast to read.
Why: joining and grouping require Spark to move matching rows to the same machine — a "shuffle" — which is the most expensive operation in distributed computing, so understanding it is the key to fast jobs. When: every join and groupBy triggers a shuffle unless avoided. Where: broadcasting a small table to every node skips the shuffle entirely, a huge win when one side is small.
from pyspark.sql import functions as F
users = spark.read.parquet("s3://lake/users/") # small dimension table
events = spark.read.parquet("s3://lake/events/") # large fact table
# Broadcast the SMALL table to every node -> no shuffle of the big one:
joined = events.join(F.broadcast(users), on="user_id", how="left")
# A plain join of two LARGE tables shuffles both — expensive but sometimes
# unavoidable. Filter and aggregate BEFORE joining to shrink the data first.Why: how Spark writes output determines how fast everyone reads it later — too few partitions underuse the cluster, too many create a "small files problem" that cripples reads. When: repartition before writing to control file count and size (aim for ~100MB–1GB files). Where: partitionBy on a query-filter column enables pruning; coalesce reduces file count without a full shuffle.
# Control output file count/size before writing:
(daily
.repartition(8, "dt") # ~8 files per dt partition
.write
.partitionBy("dt") # folder per date -> prune on read
.mode("overwrite")
.parquet("s3://lake/daily/"))
# Too many tiny files kills read performance. Aim for ~100MB-1GB per file;
# use .coalesce(n) to merge without a full shuffle.Why: when a DataFrame is reused across several actions, recomputing it each time wastes the cluster — caching keeps it in memory — and reading the query plan reveals costly shuffles before you run the job. When: cache a DataFrame used more than once; check explain() when a job is slow. Where: the plan shows Exchange (shuffle) steps, which are what to minimize.
clean = df.filter(F.col("event") == "purchase").cache() # reuse without recompute
clean.count() # materializes and caches
clean.groupBy("user_id").count().show() # reuses the cached data
# See what Spark will actually do — look for "Exchange" (a shuffle):
daily.explain()