FIT5202 · Data Processing for Big Data
Parallel Join
Week 3 of Monash FIT5202 covers parallel join — the most communication-heavy Volume operation. It contrasts the two data-partitioning strategies for join, divide-and-broadcast and disjoint (hash/range) partitioning, over the serial local-join methods (nested-loop, sort-merge, hash join), and introduces the cost model C_total = C_CPU + C_I/O + C_network in which the network (shuffle) term usually dominates. The divide-and-broadcast data-movement calculation and which-cost-dominates reasoning are common written-response items.
What this chapter covers
- 01Parallel join in two steps: data partitioning, then a local join in each partition
- 02Divide-and-broadcast: divide one table into disjoint fragments, broadcast (replicate) the other to all
- 03Disjoint partitioning (hash or range): co-partition both tables so matching keys meet on one processor
- 04Local (serial) join methods: nested-loop, sort-merge, and hash join (build on the smaller table, probe with the larger)
- 05Cost model C_total = C_CPU + C_I/O + C_network; the network/shuffle term dominates in big data
- 06Divide-and-broadcast cost phases: scan, select, broadcast (data transfer), receive, store
- 07The effect of skew on join performance and load balancing
- 08Parallel outer-join methods (ROJA, DOJA, DER) and OJSO skew optimisation (named)
Divide-and-broadcast: data movement and why the network cost dominates
- +1(a) Broadcasting means each processor sends its own fragment Sᵢ (1000 records) to the other N − 1 = 3 processors: 1000 × 3 = 3000 records sent.
- +1(b) Each processor already holds its own Sᵢ and receives the other three fragments: |S| − |Sᵢ| = 4000 − 1000 = 3000 records received. Afterwards it holds all 4000 records of S plus its own Rᵢ.
- +1(c) Across all 4 processors the broadcast moves 4 × 3000 = 12 000 record-transfers. This data-transfer term (Sᵢ/P)·(N − 1)·(mp + ml) grows with N, so the network/shuffle cost — not CPU or local I/O — is the parallel-join bottleneck in big data.
- +1Local join: each processor now joins its disjoint Rᵢ against the full S, so every R record can meet every possible S match on its own processor.
- +1Union the local results to get the global join. Because one table is fully replicated, no matching pair is ever split across processors, so no join result is missed — the trade-off for the heavy broadcast traffic.
Key terms
- Divide-and-broadcast join
- Divide one table into disjoint fragments (one per processor) and broadcast (replicate) the entire other table to every processor, then join locally. Simple and complete, but moves a lot of data.
- Disjoint-partitioning join
- Hash- or range-partition BOTH tables on the join key so matching rows land on the same processor; each processor joins only its own fragments. Moves less data than broadcast when both tables are large.
- Hash join
- A local join that builds a hash table on the smaller relation S and probes it with rows of the larger relation R. Typically the fastest serial local-join method for equijoins.
- Cost model
- The optimizer's estimate of resource use, C_total = C_CPU + C_I/O + C_network. In big data the network (shuffle) cost usually dominates, so it drives the choice of join strategy.
- Shuffle (network) cost
- The cost of moving records between processors during a parallel join or group-by. It scales with the number of processors and the data volume, and is the usual parallel-join bottleneck.
- Sort-merge join
- A local join that sorts both inputs on the join key and then merges them in a single pass. Useful when inputs are already sorted or too large to hash comfortably.
Parallel Join FAQ
When should I use divide-and-broadcast versus disjoint partitioning?
Use divide-and-broadcast when one table is small enough to replicate cheaply — you broadcast the small table and every processor joins locally with no key-matching worries. Use disjoint (hash) partitioning when both tables are large, so co-partitioning on the join key moves far less data than broadcasting a big table. The decision is a cost-model comparison dominated by network traffic.
Why is the network cost the bottleneck rather than CPU or disk?
Because parallel joins move records between processors, and that shuffle grows with both the data volume and the processor count. Local CPU and I/O are done in parallel and overlap, but the network is a shared, slower resource, so the C_network term usually dominates C_total in big-data joins.
What is the local join method inside a parallel join?
Once partitioning has co-located the rows, each processor runs an ordinary serial join on its fragments — nested-loop, sort-merge or (most often) hash join, building a hash table on the smaller side and probing with the larger. The parallelism is in the partitioning; the local join is classic single-machine work.
Can AI help me with parallel-join cost models?
Yes. Sia can walk through the divide-and-broadcast cost phases, compute the data moved for given N and table sizes, and explain why disjoint partitioning wins for two large tables — step by step, checking your working. It does not do your graded assignment, and Monash academic-integrity rules apply.
Exam move
Anchor Week 3 on the cost model C_total = C_CPU + C_I/O + C_network and the fact that the network/shuffle term dominates. Be able to compute, for divide-and-broadcast, how much data each processor sends and receives, and to argue when disjoint partitioning is cheaper. Know the three local-join methods and when hash join is fastest. This is prime written-response material for Q2–Q6, so practise a full data-movement calculation on Moodle problems through SWOTVAC rather than only reading the slides.
Working through Parallel Join in FIT5202? Sia is AskSia’s AI Information Technology tutor — ask any FIT5202 Parallel Join question and get a clear, step-by-step explanation grounded in how FIT5202 is taught and assessed. Read this chapter free, then take your hardest questions to Sia.
