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 & climate | Atmosphere in 3D cells: temp, pressure, wind | IMD, ECMWF, NOAA |
| CFD / aerodynamics | Air over a wing or car body | Boeing, F1, ISRO |
| Molecular dynamics | Positions & forces of atoms | Drug discovery |
| Astrophysics | Gas & dark matter under gravity | Cosmology |
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 strips | 10,000 | 2,000 | 20% |
| 2-D blocks | 10,000 | 400 | 4% — 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 |
|---|---|---|
| 100 | 20% | 4% |
| 1,000 | 200% | 13% |
| 10,000 | 2,000% — pure communication | 40% |
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-memory | One box, all cores share RAM & disk | To one machine's limit |
| Shared-disk | Separate CPUs, one shared storage | Storage bottlenecks |
| Shared-nothing | Nothing — each node owns its slice | Yes — 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 shard | Dead. 9 idle shards and 1 on fire. |
country | 6,000/s — India is 60% of your users | 3× over capacity, and it gets worse as you grow |
hash(user_id) | ~1,000/s — evenly spread | Comfortable. 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 shard | all 10 shards |
| "All orders in March" | all 10 shards | 1 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 |
|---|---|---|
| Partition | Split the grid | Shard the table |
| Compute | Update my cells | Scan my shard |
| Communicate | Halo exchange | Fan-out / shuffle |
| Combine | Assemble the field | Merge partials |
Same idea, different clothes. Partition → compute → communicate as little as possible → combine.
Recap & What's Next
Key Takeaways
- Scientific computing splits a grid across processors (domain decomposition) and exchanges halo cells each step.
- Monte Carlo (darts → π) is embarrassingly parallel — independent samples, near-perfect speedup.
- Parallel databases use shared-nothing nodes and sharding; queries fan out and a coordinator merges.
- Joins are cheap when co-located on the join key, else you pay for a shuffle or broadcast.
- OLAP warehouses are massively parallel because a big scan is the ideal parallel job. Partition → compute → combine.
Homework
- Explain to a non-technical friend how a weather forecast and BigQuery are "the same idea."
- Look up one real supercomputer (Fugaku/Frontier/Aurora) — its core count and its science.
- Come ready: "Word-count over a terabyte across 100 machines — how do you split and combine it?"
Next session: Big Data & Parallel Processing
Hadoop, MapReduce & Spark — "partition → compute → combine" as a button anyone can press.