Local Top N Before a Global Merge

Concepts covered: Row Numbering, Window Partitioning, Ranking

The distributed probe trades the HR table for volume. "Same pattern: the 3 biggest sales per store by revenue, but sales is 8 billion rows in Spark across 200 executors. Walk me through the shuffle." The expected wrong answer describes the window faithfully: hash every row by store_id, shuffle all 8 billion, sort, number, filter. It is correct, it is what older engines do unprompted, and the interviewer is waiting for the sentence that cuts the shuffle by a few hundred times.

Each executor can compute the top 3 per store on the rows it already holds, and only those travel. The argument fits in the room: a row outside its executor's top 3 for store S has 3 rows of S on that executor ahead of it, those 3 are ahead of it globally, so it can't be in the global top 3. Say it in that direction, because "local winners might be global winners" proves nothing. The shuffle then carries at most 200 executors times 40,000 stores times 3 rows, 24 million in place of 8 billion.

== Physical Plan ==
Filter (rn <= 3)
+- Window [row_number() AS rn], [store_id], [revenue DESC, sale_id ASC]
   +- WindowGroupLimit [store_id], [revenue DESC, sale_id ASC], row_number(), 3, Final
      +- Sort [store_id ASC, revenue DESC, sale_id ASC]
         +- Exchange hashpartitioning(store_id, 200)
            +- WindowGroupLimit [store_id], [revenue DESC, sale_id ASC], row_number(), 3, Partial
               +- Sort [store_id ASC, revenue DESC, sale_id ASC]
                  +- FileScan parquet sales

Spark 3.5 added this as the WindowGroupLimit rule: a filter like rn <= 3 directly over ROW_NUMBER gets a Partial node before the Exchange and a Final one after, and the same holds for RANK and DENSE_RANK. On earlier versions, or when the filter hides behind a UDF or a view, the plan shows a bare Exchange of all 8 billion rows, and SQL can't fix it: a window partitioned by spark_partition_id() still shuffles every row, because a window needs its partition keys co-located. The hand-written local stage lives below SQL, as a per-partition heap of 3 per store in mapPartitions, or an aggregate whose buffer holds the top 3. Ask which version runs, and look for the Partial node below the Exchange.

Rank values that break the local cut

The local cut fails when you rank by an aggregate. Ask for the top 3 products per store by total revenue: a product's revenue is split across executors, and one that is 5th on every executor can be 1st once its partials add up. A local cut needs a rank value that no other executor's rows can still change. So aggregate first, which Spark does in 2 phases of its own, then cut locally, then merge.

Two more sentences earn credit. Skew: the single-pass window sends all 400 million rows of the largest store to 1 executor, the task that runs 40 minutes while 199 finish; after the local cut that store sends at most 600 rows, so the local stage is also the skew fix. Determinism: both stages need a total tiebreaker. If 2 sales tie at 9,400 and the ORDER BY stops at revenue, which one an executor keeps depends on read order, and 2 runs over identical data return different rows.

About this lesson

This section is part of the Top N Per Group: Advanced lesson on DataDriven, the home of data engineering. Lessons cover SQL, Python, Spark, data modeling and pipeline architecture, and each section pairs its explanation with worked examples and practice you run as you go.