At Tech Fabric, working with one of our clients, we noticed that an incremental pipeline was taking about as long as a full batch run would, around 60 minutes on a 16-node cluster. We brought it down to 10–15 minutes on 8 nodes by changing how the data was filtered and joined.
The pipeline runs canonical transformations over insurance and healthcare data on Spark and Delta Lake, and several of its steps joined large DataFrames before filtering the joined result. The Spark UI and the Spark SQL execution plans showed those joins taking in substantially more data than the transformations needed, so I moved the relevant filters onto the input DataFrames.
Join first, filter later
Joining first and filtering afterward reads naturally in code, since you bring the sources together and then keep the rows the transformation needs.
Adding nodes was the obvious response. I went to the Spark UI first, because more compute speeds up the work a job does without showing whether that work was necessary, and the execution plan does show it.
What the Spark UI showed
I worked through the Spark UI and the Spark SQL execution plans. I looked at how much data the pipeline was reading, what was happening at the joins, and the processing that followed them.
The large joins were taking in more data than the transformations used, and the data flow carried that extra volume forward into the rest of the job. The clearest opportunity was to shrink what entered the joins.
Filter the inputs, then join
I restructured the transformation logic so the relevant filtering conditions applied to the input DataFrames before the joins, instead of to the joined result. Each join was now written over only the records its transformation needed. That changes the order in the code. Whether it also changes what Spark executes depends on what Catalyst was already doing with the original filters, which the next section covers.
When filtering earlier does shrink what reaches a join, Spark has less data to move and compare, whichever strategy it picks, sort-merge or broadcast.
For an inner join with a condition on one input's columns, moving the filter before the join keeps the results the same. With an outer join, a condition on the side that can come back null gives different results before and after the join, so check the join type and null handling first.
Why Catalyst matters
Catalyst, the Spark SQL optimizer, rewrites the logical plan before anything executes, so writing a filter before a join in code doesn't necessarily make Spark run it any earlier. A filter written after the join may already end up in the same place. Catalyst's PushPredicateThroughJoin rule separates a filter's conditions into those that can be evaluated on the left input, on the right input, or only on the combined result.
| Filter written after the join | Can Catalyst push it below the join? |
|---|---|
| Inner join, simple filter on one input | Usually yes |
| Inner join, deterministic predicate on both inputs | May split or rewrite it, depending on the expression |
| Left outer join | Depends on which rows the join must preserve; conditions on the preserved (left) side can generally be pushed down |
| Right outer join | Depends on which rows the join must preserve; conditions on the preserved (right) side can generally be pushed down |
| Full outer join | Depends on the predicates; a null-rejecting filter can turn it into a left, right or inner join first |
| Non-deterministic condition | Generally restricted from being pushed |
| Filter that depends on the completed join | May need to remain after the join |
Outer joins have one more path. When a filter on the side that can come back null also rejects those nulls, the EliminateOuterJoin rule can turn the outer join into an inner join before the pushdown runs.
Pushing a filter through a join is also different from data source pushdown. The first changes where the filter sits in the plan. The second hands an eligible filter to the scan so the reader can skip data, and on Delta tables Databricks uses per-file minimum and maximum statistics to skip files that can't match. A filter can sit below the join and still sit above a scan that reads most of the files, for example when its column has no collected statistics or its values are spread across every file.
Before deciding a rewrite will help, open the optimized and physical plans. They show where filters actually run, which join strategy Spark chose, and how much data reaches each operator. In our pipeline, the restructured logic processed less data, and runtime fell to about 10–15 minutes alongside the move to 8 nodes. We didn't benchmark the code change and the resize separately.
The results
| Metric | Before | After |
|---|---|---|
| Runtime | Around 60 minutes | About 10–15 minutes |
| Cluster size | 16 nodes | 8 nodes |
After the change, the pipeline completed in about 10–15 minutes on half the nodes. We observed lower compute usage and cost, but we haven't quantified the savings.
What to check before adding nodes
When a Spark job is slow, these are the places in the Spark UI worth checking before you resize anything:
- Input rows and bytes. Compare what the scans read with what the job actually needs. A large gap usually means filtering happens later than it could.
- Join operator metrics. Look at rows in versus rows out, and at the join strategy Spark chose. A big drop right after a join suggests it did work on rows nobody kept.
- Shuffle read and write. The stages around each join show how much data moved between executors.
- Memory and disk spill. Spill means a stage needed more execution memory than it had and wrote data to disk, which slows it down.
- Task duration. If a few tasks in a stage run far longer than the median, look for skew on the join key.
- Optimized and physical plans. Confirm where each filter actually runs before deciding a rewrite will help.
On this pipeline, starting with the plan turned a sizing question into a change to the transformation logic, and the job ended up faster on half the nodes. On the next slow job, I'd open the SQL / DataFrame tab before touching the cluster settings, and size compute only once the plan shows the work is necessary.
Frequently asked questions
Does filtering before a join always make Spark faster?
No. Catalyst can already move an eligible filter below an inner join, so for a simple condition on one input, moving it earlier in code may change nothing. Check the optimized and physical plans to see where the filter already runs.
Is it safe to move a filter before an outer join?
Not always. A condition on the side of an outer join that can come back null gives different results before and after the join. Check the join type and null handling, and compare the outputs, before relying on the rewrite.
Does pushing a filter below a join mean Spark reads less data?
Not necessarily. Moving a filter below a join changes where it runs in the plan, while reading less data depends on data source pushdown and, on Delta tables, on file statistics that let Databricks skip files. Check the scan node in the SQL / DataFrame tab for the filters it applied and the number of files it read.
How much did the change save in cost?
We observed lower compute usage and cost, but we haven't calculated the DBU or dollar savings. To put a number on a change like this, compare DBU usage from your billing data across comparable runs before and after it, rather than multiplying nodes by minutes, since node type and pricing tier change the result.
Where in the Spark UI should I start?
Start with the SQL / DataFrame tab for the slow job. Compare the input rows and bytes with what the job needs, then look at the join operators, shuffle read and write, spill and task durations in the stages around the joins.