explain() and query plans: practical tips and trade-offs

Why query plans matter in production

Every Spark job executes a plan before it runs a single task. Understanding explain() and query plans is the difference between a job that completes in seconds and one that crawls through petabytes of data unnecessarily. When performance degrades or a join hangs, the query plan is your first window into what Spark will do.

In this post, we'll walk through practical tips and trade-offs for reading and optimizing Spark query plans. We'll cover the tools Spark gives us, what to look for, and how to iterate on your logic to improve execution.

Understanding df.explain(mode="formatted")

The simplest way to see what Spark plans to do is call explain() on your DataFrame. Spark offers several modes; "formatted" is the most readable for humans.

from pyspark.sql import SparkSession

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

df = spark.read.parquet("s3://my-bucket/events/")
df_filtered = df.filter(df.user_id == "12345")
df_joined = df_filtered.join(
    spark.read.parquet("s3://my-bucket/users/"),
    on="user_id",
    how="inner"
)

df_joined.explain(mode="formatted")

This produces a human-readable breakdown of the logical plan, optimized plan, and physical plan. The "formatted" mode organizes the output into sections:

• Parsed Logical Plan: what your code literally says
• Analyzed Logical Plan: after Spark resolves column names and types
• Optimized Logical Plan: after Catalyst applies rule-based transformations
• Physical Plan: the actual operations Spark will execute on executors

Most of the time, you'll focus on the Physical Plan. That's where you see whether Spark is doing what you expect.

Other modes include "simple" (terse), "extended" (very verbose), and "cost" (estimates, if adaptive query execution is enabled). For day-to-day work, "formatted" strikes the right balance.

Look for scans, joins and exchanges

Three operations dominate Spark performance bottlenecks: scans (reading data), joins (combining datasets), and exchanges (shuffling data across the network). Learning to spot these in a query plan is essential.

Scans: are you reading too much?

A scan is your first interaction with persistent data. Look for operations like FileScan parquet, FileScan orc, or FileScan csv. The output will show you the path, the number of files, and sometimes the size.

If you're scanning 50 GB when you only need 5 GB, that's a red flag. This often happens when:

• Partitioning isn't being exploited (we'll cover filters next)
• You're reading the entire table instead of a date range
• A join is reading both tables in full before filtering

In a real scenario, you might scan a daily events table with millions of rows. If your filter isn't pushed down to the file level, Spark reads the entire partition, then filters in memory. That's wasteful.

Joins: what strategy is Spark using?

Joins appear as BroadcastHashJoin, SortMergeJoin, or ShuffledHashJoin in the physical plan. Each has different cost and performance characteristics:

• BroadcastHashJoin: copies the smaller table to all executors. Fast, no shuffle. Only works if the small table fits in memory (default threshold: 10 MB, configurable via spark.sql.autoBroadcastJoinThreshold).
• SortMergeJoin: sorts both tables and merges them. Reliable for large joins but involves a shuffle.
• ShuffledHashJoin: rarer; similar cost to SortMergeJoin.

If you're joining a large fact table with a small dimension table, and Spark chooses SortMergeJoin instead of broadcast, it's doing unnecessary work. Conversely, if Spark tries to broadcast a 2 GB table and your executors only have 1 GB of memory, the job will fail or spill to disk.

Here's an example where we explicitly hint Spark:

from pyspark.sql.functions import broadcast

orders = spark.read.parquet("s3://my-bucket/orders/")
customers = spark.read.parquet("s3://my-bucket/customers/")  # small, ~50 MB

# Explicitly broadcast the small table
result = orders.join(broadcast(customers), on="customer_id", how="inner")

result.explain(mode="formatted")

The physical plan should now show BroadcastHashJoin. If it doesn't, Spark may be overriding your hint because it thinks the estimate is wrong; that's a sign to check your table statistics.

Exchanges: when is Spark shuffling?

An exchange (shuffle) appears as Exchange hashpartitioning or similar in the plan. Shuffles are expensive: data crosses the network, gets serialized, written to disk, and deserialized on the other side.

Common causes of shuffles:

• Joins on columns that don't match the current partitioning
• Aggregations (groupBy) over the entire dataset
• Sorting (orderBy) for a global result
• Repartitioning (explicit .repartition())

If you see multiple shuffles in succession, it's a smell. Often you can reorganize your logic to avoid them. For example, if you filter before a join, Spark can reduce the amount of data shuffled.

Check pushed filters

A pushed filter is one that Spark applies at the file level, before bringing data into memory. This is one of the biggest wins in query optimization.

In the physical plan, look for predicates under the scan operation. If you see your filter condition nested inside the FileScan, Spark pushed it down. If you see it as a separate Filter operator above the scan, it didn't.

Example: filtering by partition column.

events = spark.read.parquet("s3://my-bucket/events/")
# Partitioned by date

# This filter will likely be pushed
result = events.filter(events.date == "2024-01-15")
result.explain(mode="formatted")

In the output, you should see something like:

FileScan parquet ... 
  PushedFilters: [IsNotNull(date), EqualTo(date, 2024-01-15)]

Now contrast with a filter that can't be pushed:

result = events.filter(events.user_id.isin(["123", "456", "789"]))
result.explain(mode="formatted")

If user_id isn't a partition column and the file format doesn't support predicate pushdown on that column (e.g., simple CSV), Spark reads the whole file and filters in memory. That's acceptable if the data is small, but across terabytes, it compounds.

Strategies to improve filter pushdown:

• Partition your data by the columns you filter on most (date, region, customer segment)
• Use Parquet or ORC; they support pushdown better than CSV or JSON
• Filter early in your pipeline, before joins or other operations
• Avoid complex expressions in filters; Spark can't always optimize them

Compare plans after changes

The real power of explain() comes from iteration. Change your code, re-run explain(), and compare. This is how you optimize in practice.

Scenario: you have a large join that's slow.

orders = spark.read.parquet("s3://my-bucket/orders/")
items = spark.read.parquet("s3://my-bucket/items/")

# Version 1: naive join
result_v1 = orders.join(items, on="order_id", how="inner")
result_v1.explain(mode="formatted")

# Version 2: filter before join to reduce shuffle
result_v2 = (
    orders.filter(orders.status == "completed")
    .join(items, on="order_id", how="inner")
)
result_v2.explain(mode="formatted")

Compare the two plans. In version 2, the filter reduces the size of the orders table before the join. The exchange (shuffle) will be smaller. The physical plan for version 2 should show fewer rows being shuffled, visible in the estimated row counts.

Another common iteration: adjusting broadcast threshold.

spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "100mb")
result = orders.join(items, on="order_id", how="inner")
result.explain(mode="formatted")
# Now check if a previously sort-merge join became broadcast

Keep a notebook or document where you store before-and-after explain() outputs. This builds intuition over time and helps you justify optimizations to the team.

Common mistakes

• Ignoring row count estimates: the numbers next to operators (e.g., rows: 1000000) are estimates. If they're wildly off, Spark's optimizer makes bad choices. Collect table statistics with ANALYZE TABLE or df.write.mode("overwrite").option("path", "...").saveAsTable().
• Calling explain() before an action: explain() doesn't trigger execution; it only shows the plan. You won't see runtime statistics until you actually run the job. Use the Spark UI for runtime metrics.
• Over-optimizing small data: if a table is 100 MB, join strategy rarely matters. Spend your effort on the big tables and the jobs that run in production daily.
• Assuming one optimal plan: Spark's optimizer improves with every release. A plan that's optimal today might not be tomorrow. Revisit slow jobs periodically.
• Missing dynamic filter pushdown: in recent Spark versions, filters from one side of a join can push down to the other. If you don't see it, it might be disabled. Check spark.sql.optimizer.dynamicPartitionPruning.enabled.

Key takeaways

• Always call explain(mode="formatted") before deploying a new job or troubleshooting a slow one.
• Focus on scans, joins, and exchanges in the physical plan—these are where time and resources go.
• Pushed filters are your friend; structure your data and queries to enable them.
• Iterate on your code by comparing explain() outputs side-by-side.
• Use the query plan to make informed decisions about broadcast thresholds, partitioning strategy, and filter placement.
• Remember that estimates in the plan are just that—estimates. Validate with actual Spark UI metrics on a production or production-like dataset.