Distributed Systems I — Scaling Out

Why one machine isn't enough

For most of this course we've reasoned about a single computer: one CPU, one block of memory, one disk. That model is enough for an astonishing amount of data work — a modern laptop can sort a few hundred million rows. But eventually you hit a wall, and it's worth being precise about what the wall is.

Remember GridDP, the running example from the systems-design thread: a platform ingesting telemetry from thousands of sensors on an electrical grid. Each sensor reports several times a second, and it adds up to roughly 150 million rows per day — call it 50 GB raw, and far more once you keep months of history for analysis. Now ask two questions:

• Can one machine store it? A year of telemetry is many terabytes. You can buy a big disk, but you can't buy an infinite one, and a single disk is a single point of failure.
• Can one machine process it fast enough? "Average voltage per substation over the last 90 days" has to read billions of rows. One CPU reading one disk has a hard ceiling on how fast that goes.

There are two ways to get more capacity. The first instinct is to scale up (vertical scaling): buy a bigger box — more RAM, more cores, faster disks. This is wonderfully simple because your code doesn't change; it's still one machine. But it runs into two walls. There's a physical wall — the biggest server money can buy is finite, and you can't keep doubling RAM forever. And there's a cost wall — high-end hardware is priced exponentially: the top-tier machine costs far more than 10× a commodity one. Past a point you pay more and more for less and less.

So we scale out (horizontal scaling): instead of one heroic machine, use many ordinary ones — a cluster of nodes — and divide the data and the work among them. Twenty cheap machines can hold and crunch what no single machine could. This is the entire premise of Hadoop, Spark, BigQuery, Snowflake, and every "big data" system you'll meet later.

That last clause is the catch, and it's what the rest of this chapter is about. Scaling out is only a win if you can actually split the problem across the machines — and some problems split far more cleanly than others.

Partitioning & sharding

The first move in any distributed system is to decide which data lives on which machine. We don't put a random scatter of rows on each node; we split the data deliberately using a partition key (in databases this is often called a shard key; the terms overlap). You pick a column, and each row's value in that column decides which node it lands on.

A common scheme is hash partitioning: take the key, hash it, and take the result modulo the number of nodes. For GridDP we might partition by sensor_id:

Now no single node holds everything, and — crucially — each node can do its share of a query at the same time. Storage and compute both spread across the cluster.

But partitioning has a failure mode that catches everyone eventually: skew. If your partition key is unevenly distributed, one node gets a mountain of data while the others sit nearly idle. Imagine partitioning GridDP by region when 80% of the sensors are in one dense city region — that region's node holds 80% of the rows, runs 80% of the work, and the whole job waits on it. You bought four machines and got the speed of barely one.

Parallel processing: the map-reduce idea

Once the data is partitioned, how do we compute over it? The foundational pattern — old enough to predate Hadoop, durable enough to underlie Spark — is map-reduce. It has two phases:

• Map — each node runs the same work on its own partition, independently and in parallel. No node needs to talk to any other.
• Reduce — the partial results from each node are combined into a final answer.

Say we want the total number of telemetry rows across GridDP. Each node counts its own slice (map); then we add the four partial counts together (reduce):

The reason this is fast: the expensive part — reading and counting billions of rows — happens on all nodes simultaneously, and only a tiny result (four numbers) has to move afterward. With four nodes you get roughly 4× the throughput.

Work like this — where each partition can be processed entirely on its own — is called embarrassingly parallel. Counting rows, filtering (WHERE voltage > 250), applying a transformation to each row, computing a per-row derived column: none of these need data from any other partition, so they scale almost perfectly. Add machines, go faster, nearly linearly.

The shuffle — where the cost lives

Here's the problem. Suppose GridDP is partitioned by sensor_id, and you ask for average voltage per region. The rows for a single region are scattered across all four nodes — because they were placed by sensor, not by region. To group by region, every region's rows must end up together on one node. The cluster has no choice but to move data between machines over the network. That movement is called the shuffle.

Why is the shuffle the slow part? Because of everything you learned in the networking chapter: the network is orders of magnitude slower than reading from local memory or even local disk. During a shuffle, potentially the entire dataset gets serialized, sent across the wire to other nodes, and deserialized again. Map phases read local data and barely touch the network; a shuffle floods it. In real Spark jobs, the shuffle is overwhelmingly the most common reason a job is slow or runs out of memory.

Replication — copies for safety and speed

Partitioning answers "where does each piece of data live?" — but it leaves a glaring problem. If Node 2 holds the only copy of its slice and Node 2's disk dies, that data is gone. With more machines, the chance that some machine fails on any given day goes up, not down. So distributed systems keep replicas: multiple copies of each partition on different nodes.

Replication buys two distinct things:

• Durability / availability — if one node dies, a copy on another node takes over and no data is lost. The system survives hardware failure, which at cluster scale is a when, not an if.
• Read scaling — many copies means many nodes can serve reads of the same data at once, multiplying read throughput.

The most common arrangement is leader/follower (also called primary/replica): one replica is the leader that accepts writes; the others are followers that copy the leader's changes and serve reads.

The fundamental tension

Step back and notice the shape of everything above. Adding machines gives you more storage and more throughput — but every benefit comes paired with a cost, and the cost is almost always the network. Partitioning spreads work, but querying across partitions forces a shuffle over the network. Replication protects you, but syncing copies sends traffic over the network. More machines means more coordination, and coordination is slow because the machines can only talk over that slow, unreliable link.

This is the law to carry forward: in distributed systems, the network is the bottleneck. A single machine moves data between CPU, memory, and disk at speeds the network can't touch. The art of distributed data is arranging your data and your queries so that as much work as possible stays local, and as little as possible has to cross the wire.

✓ Check yourself

• Can you explain the difference between scaling up and scaling out, and one wall each hits?
• What makes a good partition key — and what is skew?
• Which is embarrassingly parallel: a filter, or a group-by on a non-partition column? Why?
• Can you say in one sentence what a shuffle is and why it's slow?

SELECT customer_id, AVG(voltage)
FROM telemetry
GROUP BY customer_id;

Next

We saw that replication keeps a cluster alive but quietly raises the hardest question in the field — what happens when the copies disagree. → Distributed Systems II — Consistency & Delivery