5 PySpark Performance Patterns That Actually Matter at Scale

Pattern 1: Pre-partition on your join key

The most expensive PySpark operation is a shuffle — data moving between executors across the network. Every groupBy and join triggers a shuffle unless data is already partitioned correctly. If you’re joining on the same key repeatedly, partition once and persist.

# Triggers shuffle on EVERY join — bad
result = large_df.join(other_df, "subscriber_id")

# Partition once, reuse — good
large_partitioned = large_df     .repartition(200, "subscriber_id")     .persist(StorageLevel.MEMORY_AND_DISK)

result1 = large_partitioned.join(df_a, "subscriber_id")
result2 = large_partitioned.join(df_b, "subscriber_id")

Pattern 2: Broadcast joins for dimension tables

If one DataFrame fits in executor memory (typically under 200MB), broadcast it. This eliminates shuffle entirely. Our subscriber dimension join went from 4.2 minutes to 23 seconds on an 80M record fact table with this single change.

from pyspark.sql.functions import broadcast

result = large_fact_df.join(
    broadcast(small_dim_df),  # force broadcast
    "plan_id", "left"
)

Pattern 3: Filter and project at the source

Don’t rely on Catalyst to push filters down. Explicitly filter and select only needed columns as close to the source read as possible. Scanning 200 columns when you need 8 is wasteful — and it compounds at scale.

# Bad: scan everything, filter late
df = spark.read.parquet("s3://cdrs/")     .join(subscribers, "subscriber_id")     .filter(col("event_date") >= "2026-01-01")

# Good: filter and project immediately at read
cdrs = spark.read.parquet("s3://cdrs/")     .filter(col("event_date") >= "2026-01-01")     .select("subscriber_id", "event_type", "duration_sec", "event_date")

Pattern 4: Handle data skew with salting

Telecom data is inherently skewed — enterprise customers with millions of CDRs each, while most subscribers have hundreds. Skewed joins cause a few slow tasks that hold up the whole job, and OOM errors on hot partitions.

def salted_join(large_df, small_df, join_key, buckets=10):
    """Distribute a skewed join across salt buckets."""
    large_salted = large_df         .withColumn("salt", (rand() * buckets).cast("int"))         .withColumn("salted_key", concat(col(join_key), lit("_"), col("salt")))
    
    small_exploded = small_df         .crossJoin(spark.range(buckets).withColumnRenamed("id", "salt"))         .withColumn("salted_key", concat(col(join_key), lit("_"), col("salt")))
    
    return large_salted.join(small_exploded, "salted_key")         .drop("salt", "salted_key")

Pattern 5: OPTIMIZE + ZORDER for Delta Lake tables

Delta tables accumulate thousands of small files with streaming or incremental writes. OPTIMIZE compacts them; ZORDER co-locates related data physically. On our CDR table, this reduced average subscriber query time from 6.8s to 0.9s. The table had 142,000 files before optimisation. After: 847.

-- Run weekly or after large ingestion batches
OPTIMIZE delta.`/mnt/data/cdrs`
ZORDER BY (subscriber_id, event_date);

-- Check file count and table health
DESCRIBE DETAIL delta.`/mnt/data/cdrs`;