TL;DR
-
collect()on large DataFrames pulls all distributed data into a single driver node. At scale, this wastes memory or crashes the job entirely. - Skipping broadcast hints on small tables forces Spark into an expensive sort-merge join when a simple broadcast would eliminate the shuffle.
- Python UDFs serialize data row by row through a separate Python process, bypassing Spark's native JVM optimization. Built-in
pyspark.sql.functionsshould be the first stop, not a UDF. - Caching a DataFrame you use once wastes cluster memory. Not caching a DataFrame you recompute five times wastes compute. The decision needs to be deliberate.
- If your table is partitioned, filtering on partition columns before reading tells the storage layer to skip irrelevant data entirely. Filtering after a full scan means the I/O cost is already paid.
- Data skew means one partition processes 80 percent of the work while the rest wait. It is the most common cause of a stage that stalls at 99 percent.
- The default
spark.sql.shuffle.partitionsvalue is 200. That number fits almost no production workload. Too few partitions spill to disk; too many create scheduling overhead that adds up across hundreds of daily runs. - Recommended fix sequence: reduce I/O first, fix shuffle patterns second, clean up execution logic last.
Most teams do not have one PySpark performance problem. They have six or seven small ones running simultaneously, each adding a few minutes to jobs that run dozens of times per day.
A job that takes four minutes instead of two does not look like a crisis. But multiply that across a pipeline with 30 jobs, running six times a day, on a cluster billed by the hour, across a full month, and the number becomes worth a conversation.
The harder part is that most of these patterns look completely correct in development. They work fine on a 10,000-row sample. The cost only surfaces when the table grows to 500 million rows and the same job starts taking 45 minutes instead of three.
I would not start debugging a Spark cost problem by reviewing cluster configuration. I would start with the job code. In most cases, the waste is already in the transformation logic, and the infrastructure is simply executing it faithfully.
These are the seven patterns I see most often, and the specific reason each one is more expensive than it needs to be.
What Good PySpark Cost Hygiene Actually Looks Like
The goal is not to optimize every job. Most jobs in a pipeline process small data and the compute cost is negligible regardless of how the code is written.
The goal is to identify the jobs where these patterns compound: large tables, repeated execution across the day, joins with significant size differences between inputs, and transformations that touch every row of a multi-billion-row dataset.
In a well-tuned pipeline, the cluster reads only the data it needs, small tables never shuffle, reused DataFrames are persisted at the right storage level, shuffle partitions match actual data volume, and skewed keys are handled before they stall a stage.
When those conditions are in place, the same pipeline typically runs at a fraction of its original cost without changing the underlying business logic. The seven patterns below are what prevent that from happening.
Mistake 1: Calling collect() Without Thinking About What It Does
collect() pulls every row of a DataFrame from every executor across the cluster into the memory of a single driver node. On a dataset with millions of rows, this is almost always the wrong decision.
The driver is not designed to hold production data volumes. When you call collect() on a 50GB DataFrame, you are asking one machine to receive and store what was distributed across 20 or 30 executors for good reason.
The failure mode is usually one of two outcomes: an out-of-memory error that kills the job, or a slow network transfer that blocks the driver while the cluster sits idle waiting for it to finish.
# Expensive: transfers all data to the driver
rows = df.collect()
# Better: filter and limit before collecting anything
rows = df.filter(df.status == "failed").limit(100).collect()
If you are collecting data for downstream Python logic, ask whether that logic can be expressed as a Spark transformation instead. In most production cases it can. If you genuinely need a local result, aggregate and filter aggressively before collecting.
The rule I apply: collect() belongs in development notebooks and small lookup queries. In production code on large tables, it is almost always a signal that a step needs to be redesigned as a distributed transformation.
Once I/O patterns are addressed, the next most expensive issue is usually how joins are structured.
Mistake 2: Not Broadcasting Small Tables in Joins
When Spark joins two DataFrames without additional guidance, it defaults to a sort-merge join. Both sides get sorted and shuffled across the network so that matching keys land on the same executor. That shuffle is expensive, particularly when one side of the join is a small reference table with a few thousand rows.
If one DataFrame is small enough to fit in executor memory, Spark can broadcast a copy of it to every executor instead. The join then happens locally without any network shuffle at all.
from pyspark.sql.functions import broadcast
# Without broadcast: both DataFrames shuffle across the network
result = large_events_df.join(small_lookup_df, "product_id")
# With broadcast: small_lookup_df is sent to every executor, shuffle eliminated
result = large_events_df.join(broadcast(small_lookup_df), "product_id")
Spark will auto-broadcast tables below the spark.sql.autoBroadcastJoinThreshold (10MB by default), but it relies on statistics being available, which is not always the case. Explicit broadcast hints are more reliable.
The test I apply: if one side of a join is a dimension table, a configuration table, a lookup, or anything with fewer than a few hundred thousand rows, I broadcast it by default. The cost of sending a small table to every executor is almost always lower than shuffling a large table across the cluster.
With join strategy handled, the next source of hidden cost is often the transformation functions themselves.
Mistake 3: Writing Python UDFs Instead of Using Built-in Spark Functions
Python UDFs are flexible, but they carry a real execution cost. Spark runs on the JVM. Python runs in a separate process. To use a Python UDF, every row must be serialized from the JVM into the Python process, processed, then serialized back. This row-by-row transfer bypasses the columnar execution and optimization that makes Spark fast.
For the majority of transformation logic, pyspark.sql.functions contains native equivalents that run directly on the JVM without any Python serialization overhead.
from pyspark.sql.functions import col, upper, regexp_replace, when, coalesce
# Slow: Python UDF with row-by-row serialization
from pyspark.sql.functions import udf
from pyspark.sql.types import StringType
@udf(StringType())
def clean_name(name):
return name.strip().upper() if name else None
df = df.withColumn("clean", clean_name(col("name")))
# Fast: native function, no Python serialization
df = df.withColumn("clean", upper(col("name")))
If your logic genuinely cannot be expressed with built-in functions, use Pandas UDFs (vectorized UDFs). They process batches using Apache Arrow instead of individual rows, which is significantly faster than row-by-row Python UDFs while still allowing Python logic.
The approach I follow: check pyspark.sql.functions before writing any UDF. The built-in equivalent exists more often than most engineers expect. For anything more complex, move to a Pandas UDF before writing a standard Python UDF. Standard Python UDFs on large datasets are a last resort, not a default.
Mistake 4: Caching the Wrong DataFrames (or Not Caching the Right Ones)
Caching has a cost: it occupies cluster memory. Not caching also has a cost: Spark recomputes the entire transformation chain every time the DataFrame is referenced. The mistake is not making the decision deliberately.
If a DataFrame is referenced only once in the pipeline, caching it wastes memory with no benefit. If a DataFrame is the result of an expensive join or aggregation and is referenced five times downstream, not caching it means Spark re-executes that entire computation chain five times.
from pyspark import StorageLevel
# Cache a DataFrame you will reference multiple times
enriched_df = (
raw_df
.join(broadcast(lookup_df), "id")
.filter(col("active") == True)
)
enriched_df.persist(StorageLevel.MEMORY_AND_DISK)
# Reference it multiple times without recomputation
summary = enriched_df.groupBy("region").count()
detail = enriched_df.filter(col("value") > 1000)
report = enriched_df.join(targets_df, "region")
# Release it when finished
enriched_df.unpersist()
The storage level matters. MEMORY_ONLY is fastest but drops partitions under memory pressure, triggering recomputation. MEMORY_AND_DISK is safer for production workloads. I default to MEMORY_AND_DISK and only change it when memory profiling shows the disk fallback is hurting performance.
Mistake 5: Reading Full Tables When Partition Pruning Would Reduce the Scan
If a table is partitioned on a column such as event_date or region, filtering on that column at read time tells the storage engine to skip irrelevant partitions entirely. The I/O reduction is proportional to how much of the table you are skipping.
If you read the full table and then apply the filter as a transformation, Spark scans everything first and discards what it does not need afterward. The I/O cost is already paid before the filter runs.
# Expensive: full table scan, then filter applied as a transformation
df = spark.read.parquet("s3://data/events/").filter(col("event_date") == "2026-06-01")
# Efficient: partition directory path limits the scan at the storage layer
df = spark.read.parquet("s3://data/events/event_date=2026-06-01/")
# With Delta Lake: partition pruning and file-level data skipping work together
df = (
spark.read.format("delta")
.load("s3://data/events/")
.filter(col("event_date") == "2026-06-01")
)
Delta Lake extends this further. The transaction log stores column-level statistics for each data file, which allows Spark to skip entire files based on min/max ranges, not just partition directory structure. For teams on Databricks, this data skipping is one of the most effective cost-reduction mechanisms available, and it requires no code changes beyond reading in Delta format.
The principle applies regardless of format: filter as early as possible in the query plan. Push predicates to the read layer before any joins or transformations run.
With I/O handled, the next common cost is one that hides inside stages: data skew.
Mistake 6: Ignoring Data Skew in Joins and Aggregations
Data skew happens when one or more partition key values have far more rows than the rest. During a join or aggregation, Spark assigns rows with the same key to the same executor. When a key is heavily concentrated, one executor processes most of the data while the others finish and wait.
The symptom is specific: a stage that shows 98 or 99 percent task completion and then stalls. The few remaining tasks are the overloaded partitions, and the rest of the cluster is sitting idle until they finish.
Common causes are null keys that collapse into a single partition, a handful of extremely active user or product IDs, or any categorical column with severe cardinality imbalance.
For Spark 3.0 and later, Adaptive Query Execution handles skew joins automatically when enabled:
spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
For cases where AQE is not sufficient, or for skewed aggregations, salting is the manual approach. Add a random suffix to the skewed key, aggregate at the salted level, then aggregate again to produce the final result. This distributes the overloaded partitions across multiple executors.
Nulls in join keys are worth handling separately. If null values in the join key are not meaningful for the analysis, filter them out before the join. If they are meaningful, handle them as a distinct case rather than letting them collapse into a single overloaded partition.
Skew is one of those problems that looks fine at 10 million rows and becomes the dominant bottleneck at 1 billion rows. It is worth checking for early, before the pipeline goes to production.
Mistake 7: Never Changing spark.sql.shuffle.partitions From the Default of 200
Spark defaults to 200 shuffle partitions. That number was set as a reasonable middle ground and matches almost no real workload precisely.
On small datasets, 200 partitions means hundreds of tiny tasks where the scheduling overhead outweighs the actual computation. On large datasets, 200 partitions means each task processes far more data than it should, causing spills from memory to disk and slow stage execution.
The right number depends on your data volume and cluster configuration. A practical starting point: aim for shuffle partitions where each task handles roughly 100MB to 200MB of shuffled data.
# For a 20GB shuffle, 100 to 200 partitions is a reasonable starting range
spark.conf.set("spark.sql.shuffle.partitions", "150")
# For variable workloads, let AQE tune this at runtime
spark.conf.set("spark.sql.adaptive.enabled", "true")
# AQE coalesces small partitions and splits oversized ones after the shuffle
Adaptive Query Execution, available in Spark 3.0 and enabled by default in Databricks Runtime 7.0 and later, handles partition sizing automatically by coalescing small partitions after the shuffle completes. If your runtime supports it, enabling AQE is the single configuration change with the broadest benefit. It does not replace understanding the underlying problem, but it significantly reduces the impact of a misconfigured shuffle partition count across a diverse set of jobs.
Summary and Next Steps
Most PySpark cost problems are not infrastructure problems. They are code patterns that made sense when the dataset was small and became expensive as the data grew, with no single change large enough to trigger an investigation.
The fix sequence I would follow:
Start with I/O: enable partition pruning, check that filters are applied before joins, and confirm that tables read in Delta format are benefiting from data skipping. This reduces the volume of data the cluster handles before any transformation logic runs.
Then address shuffle: broadcast small tables, review join strategies, and enable AQE for automatic partition tuning. Shuffle is where the largest cost spikes typically live, and fixing join patterns usually has the fastest visible impact on job runtime.
Finally, clean up execution: replace Python UDFs with native functions, resolve data skew with AQE or manual salting, and make caching decisions explicit rather than leaving them to default behavior.
Each of these fixes is independent. You do not need to address all seven at once. Start with the most expensive job in your pipeline, profile it using the Spark UI stage view, and identify which pattern is driving the cost.
For teams running complex pipelines on Databricks, these decisions compound quickly. The difference between a well-structured Spark job and a poorly structured one is not just runtime. It is the difference between a pipeline that costs a few hundred dollars a month and one that costs several thousand for the same output.
Lucent Innovation works with data engineering teams to design, audit, and optimize Databricks pipelines through its data engineering and Databricks consulting services. As a Certified Databricks Partner with 1,250+ projects delivered, the team has seen most of these patterns at scale and can help identify where a pipeline's cost profile diverges from what the architecture should reasonably produce.
Top comments (0)