# Spark DataFrame Optimization and Caching: 7 Best Practices from the Data Engineering Zoomcamp

> Master Spark DataFrame optimization with 7 best practices. Learn to minimize shuffles, cache wisely, and tune configurations for peak performance from DataTalksClub. Accelerate your data pipelines.

- Repository: [DataTalksClub/data-engineering-zoomcamp](https://github.com/DataTalksClub/data-engineering-zoomcamp)
- Tags: best-practices
- Published: 2026-05-31

---

**To optimize Spark DataFrames, minimize shuffles through early filtering and projection, cache only reused datasets with explicit unpersist calls, and tune shuffle partitions and memory configurations based on cluster capacity.**

Apache Spark executes DataFrame operations lazily, building a logical plan and an optimized physical plan through the **Catalyst optimizer** before materializing results. Mastering **Spark DataFrame optimization and caching** requires deliberately guiding the optimizer and managing persisted data to prevent unnecessary resource consumption. The DataTalksClub/data-engineering-zoomcamp repository demonstrates these patterns through production-grade batch processing pipelines in the `06-batch` module.

## Minimize Shuffles

Shuffles—data exchanges across the network—are the most expensive operations in distributed computing. Reducing them is the primary lever for Spark DataFrame optimization.

### Project and Filter Early

Select only the columns you need immediately after reading data. In [`06-batch/code/06_spark_sql.py`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/06-batch/code/06_spark_sql.py), the pipeline demonstrates column projection and union operations before performing a single aggregation, keeping the data flow linear and limiting unnecessary data movement.

Apply filters as early as possible in the transformation chain. Early `.filter()` operations reduce the volume of data processed in subsequent joins and aggregations, minimizing the data that must be shuffled across the cluster.

### Use Broadcast Joins for Small Tables

When joining a large DataFrame with a small lookup table, use **broadcast joins** to eliminate a full shuffle. Spark sends the entire small table to every executor, allowing the join to occur locally.

```python
from pyspark.sql.functions import broadcast

df_large = spark.read.parquet("gs://bucket/large_dataset.parquet")
df_small = spark.read.parquet("gs://bucket/lookup_table.parquet")

df_joined = df_large.join(broadcast(df_small), "key")

```

## Optimize Partitioning Strategies

Proper partitioning aligns data distribution with your query patterns, reducing the need for expensive reshuffling.

### Repartition on Join Keys

Before performing a join, repartition both DataFrames on the join key to ensure data with the same keys resides on the same partition. This technique, demonstrated in the Zoomcamp exercises, avoids a full shuffle during the join operation:

```python
df1_repartitioned = df1.repartition("common_key")
df2_repartitioned = df2.repartition("common_key")
result = df1_repartitioned.join(df2_repartitioned, "common_key")

```

### Coalesce for Output Files

When writing final results, use `.coalesce()` to reduce the number of output files. The [`06-batch/code/06_spark_sql.py`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/06-batch/code/06_spark_sql.py) script writes results using `coalesce(1)` to produce a single Parquet file, eliminating the small-file problem that degrades downstream read performance.

```python
df.coalesce(1).write.parquet("gs://bucket/output/aggregated.parquet", mode="overwrite")

```

## Cache and Persist Strategically

Caching stores intermediate DataFrames in memory or disk to avoid recomputation, but improper use exhausts cluster resources.

### When to Cache

Call `.cache()` or `.persist()` only on DataFrames that are **materialized multiple times** in downstream transformations. A common pattern in the Zoomcamp curriculum involves caching a filtered subset of data that serves multiple analytical queries:

```python
from pyspark.sql import functions as F

filtered_df = df.filter(F.col("pickup_datetime") >= "2023-01-01")
filtered_df.cache()  # Persist for reuse across multiple aggregations

# Use filtered_df in several transformations...

result_a = filtered_df.groupBy("VendorID").count()
result_b = filtered_df.agg(F.sum("fare_amount"))

filtered_df.unpersist()  # Release memory when done

```

### Storage Level Selection and Cleanup

For large datasets that exceed memory capacity, use `StorageLevel.MEMORY_AND_DISK` to spill overflow to disk rather than recomputing. Always call `.unpersist()` when the cached DataFrame is no longer needed to free executors for subsequent stages.

## Leverage Optimizer Hints

Spark SQL supports optimizer hints to force specific execution strategies when the automatic optimizer makes suboptimal choices.

Use the **broadcast hint** to ensure small tables are broadcast even when statistics are missing:

```python
df_joined = df_large.join(broadcast(df_small), "id")

```

For fine-grained control over output partitioning, use the coalesce hint in SQL expressions:

```sql
SELECT /*+ COALESCE(4) */ * FROM dataset

```

## Reduce Wide Transformation Overhead

Wide transformations (joins, aggregations) require shuffling. Mitigate their impact by pushing computations down to the data source.

### Push Aggregations Down

Compute partial aggregates on partitioned subsets before performing a global reduction. This reduces the data volume shuffled across the network.

### Prefer Built-in Aggregations

Use `groupBy().agg()` instead of `groupBy().applyInPandas()` unless you require custom Python logic. The built-in aggregations remain within the Catalyst engine, benefiting from code generation and columnar storage optimizations.

## Tune Spark Configuration

Configuration parameters in [`06-batch/setup/config/spark-defaults.conf`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/06-batch/setup/config/spark-defaults.conf) control resource allocation and shuffle behavior.

### Shuffle Partitions

Set `spark.sql.shuffle.partitions` to a value that matches your cluster cores (typically 200 for moderate clusters, or 2-3x the number of cores for large datasets):

```python
spark.conf.set("spark.sql.shuffle.partitions", "200")

```

### Executor Memory and Cores

Align `spark.executor.memory` and `spark.executor.cores` with your cluster's node capacity. Over-allocating memory leads to garbage collection pauses, while under-allocating causes disk spilling.

## Prefer DataFrames Over RDDs

DataFrames benefit from **Catalyst** (query optimization) and **Tungsten** (code generation, columnar storage) optimizations that RDDs lack. The exercises in `06-batch/code/06_spark_sql.ipynb` use pure DataFrame APIs exclusively, demonstrating the preferred approach for performance-critical workloads.

## Complete Optimization Example

The following pattern combines projection, filtering, broadcasting, caching, and configuration tuning as implemented in the Zoomcamp repository:

```python
from pyspark.sql import SparkSession, functions as F
from pyspark.sql.functions import broadcast

spark = SparkSession.builder.appName("OptimizationDemo").getOrCreate()

# 1. Project only needed columns and filter early

df = (
    spark.read.parquet("gs://bucket/green_tripdata.parquet")
         .select("VendorID", "pickup_datetime", "fare_amount")
         .filter(F.col("pickup_datetime") >= "2023-01-01")
)

# 2. Cache for reuse in multiple queries

df.cache()

# 3. Broadcast small lookup table to avoid shuffle

zones = spark.read.parquet("gs://bucket/zone_lookup.parquet")
df_enriched = df.join(broadcast(zones), "VendorID")

# 4. Tune shuffle partitions for aggregation

spark.conf.set("spark.sql.shuffle.partitions", "100")
agg = (
    df_enriched.groupBy("VendorID")
               .agg(F.sum("fare_amount").alias("total_fare"))
)

# 5. Write result with coalesced output

agg.coalesce(1).write.parquet("gs://bucket/output/agg.parquet", mode="overwrite")

# 6. Free cached resources

df.unpersist()

```

## Summary

- **Minimize shuffles** by projecting columns early, filtering before joins, and broadcasting small tables.
- **Partition wisely** using `.repartition()` for join keys and `.coalesce()` for output file management.
- **Cache strategically** on reused DataFrames with `.cache()` or `.persist(StorageLevel.MEMORY_AND_DISK)`, and always call `.unpersist()` to release resources.
- **Tune configurations** in [`spark-defaults.conf`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/spark-defaults.conf), specifically `spark.sql.shuffle.partitions` and executor memory settings.
- **Prefer DataFrames** over RDDs to leverage Catalyst and Tungsten optimizations as shown in [`06-batch/code/06_spark_sql.py`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/06-batch/code/06_spark_sql.py).

## Frequently Asked Questions

### When should I use cache() versus persist() in Spark?

Use `.cache()` when you want the default storage level (MEMORY_AND_DISK in recent Spark versions) and the DataFrame fits comfortably in memory. Use `.persist(StorageLevel.DISK_ONLY)` or `MEMORY_AND_DISK` when working with datasets larger than available RAM to prevent out-of-memory errors. Both methods require calling `.unpersist()` when the data is no longer needed to free cluster resources.

### How do I choose the right number for spark.sql.shuffle.partitions?

Set `spark.sql.shuffle.partitions` to 2-3 times the number of CPU cores in your cluster, typically starting with 200 for moderate workloads and scaling up to 1000+ for terabyte-scale datasets. Too few partitions create large tasks that risk memory overflow, while too many generate excessive task overhead and small files. Monitor the Spark UI and adjust based on the size of your shuffle output stages.

### What is the difference between repartition() and coalesce()?

`.repartition()` triggers a full shuffle and evenly redistributes data across the specified number of partitions, making it ideal for resolving skew before joins. `.coalesce()` reduces the number of partitions without shuffling data, simply merging partitions on existing executors, which is faster but can create uneven data distribution. Use `coalesce()` only when decreasing partition counts for final output, as shown in [`06-batch/code/06_spark_sql.py`](https://github.com/DataTalksClub/data-engineering-zoomcamp/blob/main/06-batch/code/06_spark_sql.py).