If your Spark job is slow, shuffle is almost certainly involved. Shuffle happens when Spark needs to redistribute data across partitions — typically during joins, groupBys, and aggregations. Data moves across the network, gets written to disk, and read back. It is expensive by nature. These six mistakes make it catastrophically worse.
1. Too Many Small Partitions
Spark's default partition count is spark.sql.shuffle.partitions = 200. If your dataset is 500MB, 200 partitions means each partition is 2.5MB — well below the 100-200MB sweet spot. You end up with 200 tasks doing almost no work, and the scheduler overhead dominates.
# Check your partition count and sizes df.rdd.getNumPartitions() # how many df.rdd.map(lambda x: 1).sum() / df.rdd.getNumPartitions() # avg rows per partition
Fix: Tune shuffle.partitions to match your data size.
spark.conf.set("spark.sql.shuffle.partitions", "20") # Rule of thumb: data_size_bytes / 128MB, minimum 2 * num_cores
2. Too Few Large Partitions
The opposite problem: one or a few partitions that are enormous. This is usually data skew — some keys appear far more often than others. While most executors finish quickly, a few tasks running on the large partitions drag the job to a halt.
# Find skewed partitions df.groupBy(spark_partition_id()).count().orderBy("count", ascending=False).show(10)
Fix: Salt your join key to distribute the skewed key across multiple partitions.
import pyspark.sql.functions as F # Add a random salt 0-9 to the skewed side df_large = df_large.withColumn("salt", (F.rand() * 10).cast("int")) df_large = df_large.withColumn("salted_key", F.concat(F.col("key"), F.lit("_"), F.col("salt"))) # Explode the small side to match all salts df_small = df_small.withColumn("salt", F.explode(F.array([F.lit(i) for i in range(10)]))) df_small = df_small.withColumn("salted_key", F.concat(F.col("key"), F.lit("_"), F.col("salt"))) result = df_large.join(df_small, "salted_key")
3. Joining Large Datasets Without Broadcast
When you join a large table with a small table, Spark by default shuffles both sides across the network. If the small table fits in memory, this is completely unnecessary.
# Spark shuffles both sides — slow result = large_df.join(small_df, "user_id")
# Broadcast the small table — stays in executor memory from pyspark.sql.functions import broadcast result = large_df.join(broadcast(small_df), "user_id")
Spark will broadcast tables smaller than spark.sql.autoBroadcastJoinThreshold (default 10MB) automatically. For tables up to a few hundred MB, set this higher or use broadcast() explicitly.
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "100mb")
4. groupByKey Instead of reduceByKey
In RDD-based code, groupByKey shuffles all values across the network first, then reduces. reduceByKey reduces locally on each partition first, then shuffles the smaller intermediate result. For large datasets the difference is dramatic.
# Shuffles ALL values across the network word_counts = rdd.groupByKey().mapValues(sum) # Reduces locally first, shuffles much less data word_counts = rdd.reduceByKey(lambda a, b: a + b)
In DataFrame API, use groupBy().agg() — Spark handles the optimisation automatically with Tungsten.
5. Collecting Large Results to the Driver
collect() pulls the entire dataset from all executors into the driver's memory. On a large dataset, this causes an OOM on the driver and stalls the entire job.
# Kills the driver on large data all_rows = df.collect() for row in all_rows: process(row)
Fix: Process data on the cluster, not on the driver.
# Process in parallel on the cluster df.foreach(process_row) # Or write output to storage and read selectively df.write.parquet("s3://bucket/output/") # If you truly need to collect, filter aggressively first df.filter("status = 'failed'").limit(1000).collect()
6. Caching at the Wrong Storage Level
Calling cache() on a DataFrame stores it as deserialized Java objects in memory — fast to access but uses significantly more memory than the raw data. On memory-constrained clusters, Spark evicts cached data and recomputes it, which can be slower than not caching at all.
# Default — uses more memory than necessary df.cache() # Better options from pyspark import StorageLevel # Serialized in memory — smaller, slightly slower to deserialize df.persist(StorageLevel.MEMORY_ONLY_SER) # Spill to disk if memory pressure — safer on large datasets df.persist(StorageLevel.MEMORY_AND_DISK) # For datasets that are read infrequently df.persist(StorageLevel.DISK_ONLY)
Only cache DataFrames that are used more than once in the same job. Caching a DataFrame used once wastes memory and slows things down.
The Underlying Principle
Every shuffle mistake comes down to moving more data than necessary across the network. The goal is always: reduce what crosses the wire. Smaller partitions, broadcast joins, local pre-aggregation, and appropriate caching are all ways of achieving the same thing.
When a job is slow, open the Spark UI, find the longest stage, and look at the shuffle read/write columns. That number tells you where to focus.