Unit V — Applications of Parallel Computing

Scientific Computing & Parallel Databases

Session 13 • 2311CSC501J — Parallel Processing

What You'll Learn

  • How simulations split the world across processors
  • Halo exchange & Monte Carlo methods
  • Sharding & shared-nothing databases
  • Parallel queries, joins & why OLAP scales

"A weather model and a data warehouse are the same idea in different clothes: partition, compute locally, combine."

— The through-line of this course

Simulating the World

You can't put a hurricane in a lab. So you simulate it: chop the physical world into a grid of tiny cells, store each cell's state, and step time forward — every cell updates from physics and its neighbors.

Domain What's on the grid Who uses it
Weather & climateAtmosphere in 3D cells: temp, pressure, windIMD, ECMWF, NOAA
CFD / aerodynamicsAir over a wing or car bodyBoeing, F1, ISRO
Molecular dynamicsPositions & forces of atomsDrug discovery
AstrophysicsGas & dark matter under gravityCosmology

Billions of cell-updates per step, thousands of steps. One machine takes weeks — but tomorrow's forecast can't wait. This is why supercomputers were invented.

Domain Decomposition

Split the grid across processors — each owns a block of space and simulates only its own cells. This is Session 07's Partition step, made physical.

The atmosphere over India, split across 4 processors

   +---------+---------+
   |   P0    |   P1    |     Each processor owns a
   | (NW)    | (NE)    |     rectangular block of cells
   +---------+---------+     and simulates only those.
   |   P2    |   P3    |
   | (SW)    | (SE)    |
   +---------+---------+

The problem: to update a cell you need its neighbors' values — but the cells on the edge of P0's block have neighbors living on P1, P2, and P3. So they must talk.

Halo Cells & the Trade-off

Each processor keeps a one-cell border copy of its neighbors' edges — the halo (ghost cells). Every step: exchange edges, then update.

P0's block, with a halo border received from neighbors

   . . . . . . .     ".":  halo cells — copies of neighbors'
   . # # # # # .           edges, refreshed every step
   . # # # # # .     "#":  P0's own cells, which P0 updates
   . # # # # # .
   . # # # # # .
   . . . . . . .

Computation

Local to each block — scales beautifully.

Halo exchange

Communication — the enemy (Session 04).

Want fat blocks, thin seams. A bigger block has more interior work per unit of border — that's PCAM's Agglomeration step (Session 07). Don't be chatty.

Worked Example: The Halo Tax

A 1000×1000 weather grid on 100 processors. Every processor does the same amount of work either way — 10,000 cells. The only question is the shape of its piece.

  1-D STRIPS                          2-D BLOCKS
  each proc: 10 x 1000                each proc: 100 x 100

  +--------------------------+        +------+------+------+
  |========= proc 0 =========|        | p0   | p1   | p2   |
  |========= proc 1 =========|        +------+------+------+
  |========= proc 2 =========|        | p3   | p4   | p5   |
  |           ...            |        +------+------+------+

  halo = 2 edges x 1000 cells         halo = 4 edges x 100 cells
       = 2000 cells exchanged              = 400 cells exchanged
Layout Cells computed Cells communicated Communication overhead
1-D strips10,0002,00020%
2-D blocks10,0004004% — 5× less

And it gets worse as you scale. The overhead formulas:

1-D strips:  comm/compute  =  2p / N          grows LINEARLY with p
2-D blocks:  comm/compute  =  4√p / N         grows with the SQUARE ROOT
Processors, N=1000 1-D overhead 2-D overhead
10020%4%
1,000200%13%
10,0002,000% — pure communication40%

At 10,000 processors the 1-D version spends twenty times more effort communicating than computing. Adding processors makes it slower. The 2-D version is still doing useful work.

This is the surface-to-volume principle, and it is everywhere. Work scales with a region's volume; communication scales with its surface. Compact shapes have less surface per unit volume — which is why cells are round, why animals in cold climates are stocky, and why supercomputers decompose 3-D simulations into cubes rather than slabs.

The design rule: make each processor's piece as compact as possible. Same work, less talking, and it's free — you just index the array differently.

One practical trick real codes use: exchange halos with non-blocking sends (Session 11), then compute the interior cells — which need no halo data — while the messages are still in flight. Done well, the communication becomes invisible.

Monte Carlo: π from Random Darts

Some science needs no neighbors at all. Monte Carlo answers questions by averaging many independent random samples.

   +-------------------+     Square area  = (2r)^2 = 4r^2
   |    . . * . . *    |     Circle area  = π r^2
   |   . ( * * * ) .   |
   |  .  (* * * *) .*  |     inside / total  ->  πr^2 / 4r^2  =  π/4
   |   . ( * * * ) .   |
   |    . * . . * .    |     so   π  ≈  4 × inside / total
   +-------------------+

Every dart is independent — split 10M darts across 8 cores, each throws 1.25M, then just sum the inside-counts (one reduction, Session 10). Near-perfect linear speedup.

Live demo: examples/01-monte-carlo-pi.html — watch π appear out of randomness.

Two Poles of Parallelism

Tightly coupled

Weather model.

Constant halo exchange between neighbors every step. Communication-bound. Hard.

Embarrassingly parallel

Monte Carlo.

Samples never talk. Almost no serial fraction → near-linear speedup. Easy.

Embarrassingly parallel = little or no communication between workers. Rendering movie frames, resizing a million images, grading 100 exams, hashing files — the jobs you love, because they just shard.

Callback to Amdahl (Session 08): almost no serial fraction is exactly why these scale near-perfectly.

Parallel Databases: One Box Isn't Enough

A table with two billion rows won't scan — or even fit — on one machine. Scale up (bigger box, has a ceiling) or scale out (more boxes, no ceiling). Parallel databases scale out.

Architecture What's shared Scales?
Shared-memoryOne box, all cores share RAM & diskTo one machine's limit
Shared-diskSeparate CPUs, one shared storageStorage bottlenecks
Shared-nothingNothing — each node owns its sliceYes — add nodes

Shared-nothing: each node has its own CPU, RAM, disk; nodes talk only over the network. No shared resource to contend on — the distributed-memory model (Session 02) that makes MPI & databases cousins.

Sharding & the Shard Key

Sharding = splitting a table's rows across nodes (horizontal partitioning). Pick a shard key and a rule.

users table sharded by hash(user_id) % 3

   Shard 0          Shard 1          Shard 2
   +----------+     +----------+     +----------+
   | u3  u6   |     | u1  u4   |     | u2  u5   |
   | u9  u12  |     | u7  u10  |     | u8  u11  |
   +----------+     +----------+     +----------+
   Node A           Node B           Node C

Hash sharding

Even spread, no hot spots. Great for key lookups, bad for range scans.

Range sharding

A–F, G–M… Great for ranges, risks a hot shard.

War story: shard by region and 80% of rows land in one region → one shard does all the work while the rest idle. That's load imbalance (Session 09) in a database costume. Pick a key that spreads evenly.

Worked Example: The Shard Key Decides Everything

An orders table across 10 shards, taking 10,000 writes per second. Each shard can handle 2,000/s. Three candidate shard keys.

Shard key Busiest shard gets Verdict
created_at (the date)10,000/s — every write today lands on one shardDead. 9 idle shards and 1 on fire.
country6,000/s — India is 60% of your users3× over capacity, and it gets worse as you grow
hash(user_id)~1,000/s — evenly spreadComfortable. Room to double.

Sharding by date is the most tempting and the most catastrophic. It looks perfectly balanced — if you have three years of data, each shard holds roughly a tenth of it. Then you notice that all writes are for today, so 100% of your write traffic hits one shard while nine sit idle. You paid for 10 machines and bought the capacity of one.

This has a name — a hotspot — and it is Session 09's load-imbalance problem, at the storage layer. Balanced data is not balanced load.

But hash(user_id) is not free either. Look at what each key does to your queries:

Query Sharded by hash(user_id) Sharded by created_at
"All orders for user 42"1 shardall 10 shards
"All orders in March"all 10 shards1 shard

You cannot have both. Every shard key makes one access pattern a single-shard lookup and every other one a scatter-gather across the whole cluster.

So the shard key is chosen by your dominant query, not by what looks tidy. Pick the key that (a) spreads writes evenly and (b) makes your most frequent query hit one shard. If those two fight, you're looking at a second copy of the data organised the other way — which is exactly why analytics runs in a separate warehouse.

And it is close to irreversible: re-sharding a live petabyte-scale table is a multi-week project. This is one of the highest-stakes decisions in system design, and it's made on day one, usually by someone who has been at the company a month.

Parallel Query Execution

A query fans out to every shard at once; each scans only its slice in parallel; a coordinator merges the partials.

   SELECT COUNT(*) FROM events WHERE country = 'IN'

                    Coordinator
                   /     |      \      (1) fan-out to all shards
              Shard0   Shard1   Shard2
               scan     scan     scan  (2) each scans in parallel
               =12M     =15M     =11M
                   \     |      /
                    Coordinator          (3) merge: 12M+15M+11M
                        = 38M            = 38M, one answer

This is fan-out / fan-in — the same scatter/gather as MPI (Session 11) and the map-reduce of Session 14. COUNT/SUM/MIN/MAX are reductions.

Gotcha: you can't average the averages — each shard returns SUM and COUNT; divide at the end.

Live demo: examples/02-database-sharding.html

Parallel Joins — the Hard One

A row on shard 0 may need to match a row on shard 2. How do you join tables scattered across machines?

Co-located join

Both tables sharded on the join key → matches already on the same node. Zero network. The dream.

Shuffle join

Not co-partitioned → re-send data over the network, re-partitioned by join key. Expensive.

Broadcast join

One table tiny? Copy it to every node instead of shuffling the big one.

This is why the shard-key choice matters so much. Shard both tables on the join key and a slow report goes from minutes to seconds — a real lever a CTO pulls.

Worked Example: Two Ways to Join, 1000× Apart

Join a 1 TB orders table to a 10 MB countries table, on 100 nodes. Both tables are sharded by something unrelated to the join key.

Shuffle join — the general method

Repartition both tables by the join key so matching rows land on the same node, then join locally.

network traffic
  = 1 TB + 10 MB
  ~ 1 TB across the network

At a realistic 10 GB/s aggregate: ~100 seconds of pure shuffling before a single row is joined.

Broadcast join — when one side is small

Send the whole 10 MB table to every node. The 1 TB table never moves at all.

network traffic
  = 10 MB x 100 nodes
  = 1 GB across the network

At the same 10 GB/s: ~0.1 seconds. Then every node joins its own local slice with zero coordination.

1,000× less traffic, same answer. And the reason is one sentence you've now heard in four different units: move the small thing, not the big thing. Session 04 called it computation-to-communication ratio. Session 11 called it "don't scatter data that's already in place." Unit V will call it "move the code to the data."

Your query engine already does this — if you let it. Spark broadcasts automatically when it believes one side is under spark.sql.autoBroadcastJoinThreshold (10 MB by default). Postgres, Snowflake and BigQuery all have equivalents.

The catch: it needs accurate statistics to know the table is small. Stale statistics → it picks the shuffle → your five-second query takes two minutes and nobody can explain why. "Run ANALYZE" is the unglamorous fix behind an enormous number of production incidents.

When neither side is small, you're stuck with the shuffle — unless both tables happen to be sharded by the join key already, in which case matching rows are already co-located and there is nothing to move at all. That is called a co-located join, and engineering for it is one of the strongest reasons to pick a particular shard key on the previous slide.

OLAP Warehouses & the Big Picture

OLTP

Live app DB. Millions of tiny reads/writes. Optimized for latency per transaction.

OLAP

Warehouse (BigQuery/Snowflake/Redshift). Scan-everything queries. Optimized for throughput.

A big scan is the perfect parallel workload — split a trillion rows across a thousand machines, scan in parallel, merge. That's Gustafson (Session 08) in production: bigger data in the same time.

Step Science Database
PartitionSplit the gridShard the table
ComputeUpdate my cellsScan my shard
CommunicateHalo exchangeFan-out / shuffle
CombineAssemble the fieldMerge partials

Same idea, different clothes. Partition → compute → communicate as little as possible → combine.

Recap & What's Next

Key Takeaways

Homework

Next session: Big Data & Parallel Processing

Hadoop, MapReduce & Spark — "partition → compute → combine" as a button anyone can press.