Optimize This Join

Concepts covered: Partitioning Strategies

The join question hands you 2 table sizes and asks which strategy you want and why. 'Orders is 2 TB, products is 40 MB, the join on product_id takes 50 minutes, what would you do?' The answer the interviewer wants names the strategy the planner picked by default, the strategy you would force, and the bytes that each one moves across the network. Spark's 3 join strategies are 3 answers to the same problem: rows with equal keys live on different executors and must meet somewhere.

Sort-merge join shuffles both sides by the join key into the same number of partitions. It then sorts each partition and walks the 2 sorted lists together. Every byte of both tables crosses the network once, and the sort needs memory or it spills. The planner picks it whenever both sides exceed the broadcast threshold. Broadcast hash join leaves the large side where it is. The driver collects the small side, ships a copy to every executor, and each task builds a hash table from that copy and probes it with its own partition of the large side. The large side moves zero bytes; the small side moves once per executor. Shuffle hash join is the middle path: shuffle both sides, then build a hash table from the smaller partition instead of sorting. It saves the sort but needs the smaller side's partition to fit in memory, and the planner prefers sort-merge over it unless spark.sql.join.preferSortMergeJoin is turned off.

Work it on 3 tables

Take orders at 2 TB, products at 40 MB and customers at 30 GB, on a cluster of 48 executors. For orders to products the planner sees 40 MB, above its default threshold, and picks sort-merge: 2 TB of orders is hashed and written to local disk, then fetched and sorted, so that it can meet 40 MB of products. Forcing a broadcast instead ships 40 MB to each of the 48 executors, 1.9 GB of network in total, and orders stays in place. The 50 minutes becomes the time to scan 2 TB and probe a hash table, a few minutes on this cluster. You get there by raising the threshold to 50 MB or by hinting the one join, and the hint is the safer answer because it changes 1 plan rather than every plan in the session.

from pyspark.sql.functions import broadcast

enriched = orders.join(broadcast(products), 'product_id')

spark.conf.set('spark.sql.autoBroadcastJoinThreshold', 50 * 1024 * 1024)

Orders to customers flips the arithmetic. Broadcasting 30 GB means the driver materializes 30 GB in its heap, then 48 executors each receive and hold 30 GB, with 1.4 TB crossing the network and every executor spending 30 GB of memory on a hash table before a task runs. Sort-merge moves 2 TB plus 30 GB once and holds 1 partition at a time. Sort-merge wins, and the detail worth adding is why the broadcast fails in practice: the driver's collect of the small side is bounded by spark.driver.maxResultSize, and Spark refuses any broadcast over 8 GB. The join cost here is the shuffle of orders, and if orders joins customers every night, the way to remove that shuffle is bucketing.

Bucketing: pay the shuffle once at write time

Write both tables bucketed by customer_id into the same number of buckets, say 1,024, and sorted within each bucket. Bucket 17 of orders and bucket 17 of customers then hold the same set of customer_id values, because both hash the key the same way into 1,024 buckets, so the planner reads matching buckets into the same task and runs the merge with no exchange in the plan. The shuffle still happens, once, when each table is written; every nightly join afterwards reads 2 TB and shuffles none of it. The conditions are strict: same key and same bucket count on both sides, recorded in the catalog so the planner can see them. A bucketed table rewritten by a job that forgot the clause loses the property silently, and the next night's plan shows the exchange again.

About this lesson

This section is part of the The Spark Deep Dive: Intermediate 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.