Apache Spark Data Engineering: DataFrames, Optimization, and Production Best Practices

Apache Spark has become the de facto standard for large-scale data processing. Whether you are building ETL pipelines, running analytical workloads, or training machine learning models, mastering Spark internals and optimization techniques separates performant pipelines from costly failures. This guide covers DataFrame optimization, broadcast variables, partitioning strategies, and write tuning at production scale.

Spark Architecture Refresher

Before diving into optimization, understanding Spark execution model is essential.

Driver and Executors: The driver process runs your application code, creates the SparkContext, and coordinates executors. Executors are JVM processes on worker nodes that run tasks and cache data.

DAG and Stages: Spark builds a Directed Acyclic Graph (DAG) of transformations. Wide transformations (shuffle operations like groupBy, join, repartition) create stage boundaries. Minimizing shuffles is the single biggest lever for performance.

Lazy Evaluation: Transformations are lazy — they build a logical plan. Only actions (count, show, write) trigger execution. This allows the Catalyst optimizer to reorder, prune, and fuse operations.

DataFrame API Fundamentals

Prefer DataFrames and Datasets over RDDs. They benefit from Catalyst optimization and Tungsten memory management.

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import StructType, StructField, StringType, LongType, TimestampType

spark = SparkSession.builder \
    .appName("DataEngineering") \
    .config("spark.sql.adaptive.enabled", "true") \
    .config("spark.sql.adaptive.coalescePartitions.enabled", "true") \
    .getOrCreate()

schema = StructType([
    StructField("event_id", StringType(), False),
    StructField("user_id", LongType(), True),
    StructField("event_type", StringType(), True),
    StructField("occurred_at", TimestampType(), True),
    StructField("properties", StringType(), True),
])

df = spark.read \
    .schema(schema) \
    .parquet("s3://data-lake/events/year=2026/month=05/")

Always define schemas explicitly — schema inference reads the entire dataset and doubles your I/O costs.

Catalyst Optimizer and Predicate Pushdown

The Catalyst optimizer applies logical and physical plan optimizations automatically, but you can guide it.

Predicate pushdown: Filter early to minimize data scanned.

# Good: filter before join
events_filtered = df.filter(
    (F.col("event_type") == "purchase") &
    (F.col("occurred_at") >= "2026-01-01")
)

users = spark.read.parquet("s3://data-lake/users/")
result = events_filtered.join(users, "user_id", "left")

Column pruning: Select only needed columns before expensive operations.

events_slim = df.select("event_id", "user_id", "event_type", "occurred_at")

Use df.explain(True) to inspect the physical plan and verify pushdowns are applied.

Broadcast Variables and Broadcast Joins

When joining a large table with a small lookup table (< 10 MB by default, configurable), Spark can broadcast the small table to all executors, eliminating shuffle.

# Set broadcast threshold
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", str(50 * 1024 * 1024))  # 50 MB

# Explicit broadcast hint
from pyspark.sql.functions import broadcast

country_codes = spark.read.parquet("s3://data-lake/dims/country_codes/")
result = large_events.join(broadcast(country_codes), "country_code", "left")

For non-DataFrame use cases, broadcast variables cache read-only data on all executors:

currency_map = {"USD": 1.0, "EUR": 1.08, "GBP": 1.26}
bc_currency = spark.sparkContext.broadcast(currency_map)

def convert_to_usd(amount, currency):
    rate = bc_currency.value.get(currency, 1.0)
    return float(amount) * rate

Partition Strategies

Partitioning is the most impactful performance lever. Poor partitioning causes data skew or excessive overhead.

# Check current partition count
print(df.rdd.getNumPartitions())

# repartition: full shuffle, use to increase partitions or balance skewed data
df_balanced = df.repartition(200, "user_id")

# coalesce: no shuffle, only reduces partition count
df_small = df_balanced.coalesce(50)

Target 128–256 MB per partition for batch workloads.

Salting skewed keys: When a few keys dominate, salt the key to spread load across partitions.

from pyspark.sql.functions import explode, array

# Add random salt 0-9 to the large table
df_salted = df.withColumn("salt", (F.rand() * 10).cast("int")) \
              .withColumn("user_id_salted",
                          F.concat(F.col("user_id").cast("string"),
                                   F.lit("_"),
                                   F.col("salt").cast("string")))

# Explode lookup table with the same salt range
salts = array([F.lit(i) for i in range(10)])
lookup_exploded = lookup_df \
    .withColumn("salt", explode(salts)) \
    .withColumn("user_id_salted",
                F.concat(F.col("user_id").cast("string"),
                         F.lit("_"),
                         F.col("salt").cast("string")))

result = df_salted.join(lookup_exploded, "user_id_salted").drop("salt", "user_id_salted")

Adaptive Query Execution (AQE)

Spark 3.0+ introduced AQE, which re-optimizes plans at runtime using actual partition statistics.

spark.conf.set("spark.sql.adaptive.enabled", "true")
spark.conf.set("spark.sql.adaptive.coalescePartitions.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.enabled", "true")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionFactor", "5")
spark.conf.set("spark.sql.adaptive.skewJoin.skewedPartitionThresholdInBytes", "256MB")

AQE automatically coalesces small post-shuffle partitions, converts sort-merge joins to broadcast joins when runtime sizes allow, and handles skewed join partitions.

Caching and Persistence

Cache intermediate DataFrames that are reused multiple times in the same job.

from pyspark import StorageLevel

expensive_df.persist(StorageLevel.MEMORY_AND_DISK_SER)
expensive_df.count()  # materialize immediately

# ... use expensive_df multiple times ...

expensive_df.unpersist()  # release when done

Storage level guidance:

• MEMORY_ONLY: Fastest; recomputes on OOM (no disk spill)
• MEMORY_AND_DISK: Safer for large datasets
• DISK_ONLY: Use when memory is scarce and recompute is expensive

Window Functions for Analytics

from pyspark.sql.window import Window

w = Window.partitionBy("user_id").orderBy("occurred_at")

df_analytics = df \
    .withColumn("row_num", F.row_number().over(w)) \
    .withColumn("prev_event", F.lag("event_type", 1).over(w)) \
    .withColumn("running_total",
                F.sum("revenue").over(
                    w.rowsBetween(Window.unboundedPreceding, Window.currentRow)))

# 7-day rolling average keyed on epoch seconds
rolling_w = Window.partitionBy("user_id") \
    .orderBy(F.col("occurred_at").cast("long")) \
    .rangeBetween(-7 * 86400, 0)

df_rolling = df.withColumn("rolling_7d_avg", F.avg("revenue").over(rolling_w))

Write Optimization

Partitioned writes — enable partition pruning on downstream reads:

df.write \
    .mode("overwrite") \
    .partitionBy("year", "month", "day") \
    .parquet("s3://data-lake/output/events/")

Control output file size — avoid the small-file problem:

# Repartition before write
df.repartition(100).write.mode("overwrite").parquet("s3://data-lake/output/")

# Or cap records per file
df.write.option("maxRecordsPerFile", 1_000_000).parquet("s3://data-lake/output/")

Dynamic partition overwrite — update only touched partitions:

spark.conf.set("spark.sql.sources.partitionOverwriteMode", "dynamic")

df.write.mode("overwrite").partitionBy("date").parquet("s3://data-lake/output/events/")

Monitoring and Debugging

Key Spark UI metrics to investigate:

• GC time > 10% of task time → memory pressure, tune executor memory or reduce dataset size
• Shuffle spill to disk → insufficient executor memory for the shuffle buffer
• Task duration variance > 3x → data skew, apply salting or AQE skew join

Custom error tracking with accumulators:

error_count = spark.sparkContext.accumulator(0)

def safe_parse(row):
    try:
        return parse(row)
    except Exception:
        error_count.add(1)
        return None

cleaned = df.rdd.map(safe_parse).filter(lambda x: x is not None).toDF(schema)
cleaned.count()  # trigger execution
print(f"Parse errors: {error_count.value}")

Production Checklist

Before deploying a Spark job:

1. Explicit schema defined (not inferred)
2. Filters applied before joins
3. Small dimension tables broadcast; join order considers sizes
4. Partition count appropriate (128–256 MB target per partition)
5. Skewed keys salted or AQE skew join enabled
6. Reused DataFrames cached and unpersisted after use
7. Output partitioned with controlled file sizes
8. Dynamic partition overwrite enabled for partial updates

Conclusion

Mastering Apache Spark for data engineering requires understanding the execution model, using DataFrames with explicit schemas, applying broadcast joins for dimension lookups, designing partition strategies to avoid skew, leveraging AQE for runtime optimization, caching strategically, and writing with controlled file sizes. Combined with Spark UI monitoring and structured error handling, these practices transform brittle scripts into reliable, cost-efficient data pipelines at any scale.