6. Distributed systems: the physics of many machines

Distribute only when one machine cannot hold the data, cannot process it fast enough, or must not be a single point of failure. In exchange you inherit partial failure, network delay, and the absence of a shared clock, and every distributed data tool is a strategy for living with those three.

Scale up before you scale out

A single modern server can have well over 100 cores and terabytes of RAM. Single-node engines such as DuckDB and Polars process hundreds of gigabytes on one machine faster than a small cluster, because they pay no network cost. Ask “does this really not fit on one big machine?” before reaching for Spark.

You scale out for four reasons: capacity, throughput through parallelism, fault tolerance through copies, and lower latency by placing data near users.

Partitioning (sharding)

Partitioning splits a dataset so each machine owns part of it.

Strategy	How	Good at	Risk
By key range	Users A–F on node 1, G–M on node 2	Range scans	Hot spots when many keys share a range
By hash of key	hash(user_id) mod N	Even spread	Range queries must ask every node
By time	One partition per day or hour	Event data, dropping old data	Today’s partition takes all the writes

Skew is when one partition gets far more data or traffic than the rest, such as one celebrity account. The whole job waits for the slowest partition, so skew turns 100 machines into one. Consistent hashing lets you add or remove nodes while moving only a small fraction of keys.

The same idea appears in lakehouse tables as directory partitioning (date=2026-10-30/). The principle is identical: arrange data so a query touches only the parts it needs.

Replication

Replication keeps copies of the same data on several machines.

• Single leader: one node accepts writes and ships its log to followers. Reading from a follower may return stale data; that delay is replication lag.
• Multi-leader: several nodes accept writes, so conflicting writes must be resolved.
• Leaderless: clients write to and read from several replicas. With N replicas, W write acknowledgements and R read replies, choosing W + R > N guarantees every read overlaps the latest write.

Synchronous replication waits for copies before confirming a write: safer, slower. Asynchronous confirms first and copies later: faster, but a crash can lose recent writes.

Consistency models

• Strong (linearizable): the system behaves as if there were one copy. Every read sees the latest write.
• Eventual: if writes stop, replicas converge. Meanwhile, readers may see older values.
• Read-your-writes: you always see your own updates, even if others do not yet.

CAP says that during a network partition a system must choose between consistency and availability. PACELC adds the everyday case: even with no partition, you trade latency against consistency. Ask any data store which side it picks.

Consensus

Consensus algorithms such as Raft and Paxos let machines agree on one value, like who the leader is, despite failures. They power etcd, ZooKeeper and Kafka’s KRaft mode. You will rarely implement one, but remember the lesson: coordination is slow and fragile, so good systems minimize how often machines must agree.

Failure and time

• Partial failure: some parts work while others do not. A request that times out may have failed, or may have succeeded with the reply lost. You cannot tell.
• Retries create duplicates: because of that uncertainty, the safe reaction is to retry, which means work may happen twice.
• Clocks lie: machine clocks drift and jump. Never order events from different machines by wall-clock time alone; use sequence numbers, log offsets, or event IDs.

The classic “fallacies of distributed computing” are worth memorizing: the network is not reliable, latency is not zero, bandwidth is not infinite, and topology changes.

Idempotency: the most important practical principle

An operation is idempotent if doing it twice has the same effect as doing it once. Since retries are inevitable, every pipeline step should be idempotent. Then recovery from any failure is simply “run it again”.

Techniques you will use daily:

1. Deterministic outputs: the run for 2026-10-30 always writes to the same location and overwrites it, rather than appending.
2. Upsert by key: MERGE on a natural key, so a repeated row replaces itself instead of duplicating.
3. Unique event IDs: consumers skip IDs they have already processed.
4. Stage, then commit: write to a temporary location, then publish atomically, so readers never see half a result.
5. Idempotency keys on API calls: the server recognizes a retried request and does not repeat the side effect.

Delivery semantics

Guarantee	Meaning	How achieved	Cost
At most once	Never duplicated, may be lost	Send without retrying	Data loss
At least once	Never lost, may be duplicated	Retry until acknowledged	Duplicates
Exactly once (effectively once)	Each record’s effect happens once	At-least-once delivery plus idempotent or transactional processing	Complexity, some latency

Exactly-once is an end-to-end property of the whole pipeline, never a feature of one component. When a vendor claims it, ask: exactly once from where to where?

Backpressure

When a producer is faster than its consumer, something must give: buffer (until memory runs out), drop data, or slow the producer down. Backpressure is the mechanism that slows the producer. Systems without it fail suddenly under load.

Why this matters in AI work

• An embedding job that retries a failed batch writes duplicate vectors, unless it upserts by chunk ID.
• A vector database may acknowledge an insert before the vector is searchable; that is eventual consistency.
• Pipelines calling LLM APIs hit rate limits and timeouts, so they need retries with exponential backoff, and idempotent writes of the results.
• Distributed training needs each worker to read a distinct shard of data, which is partitioning again.

Questions to ask

• Does this genuinely need more than one machine?
• How is the data partitioned, and where could skew appear?
• If this step runs twice, what happens?
• What does “exactly once” mean here, end to end?
• When the network splits, does this system stay available or stay consistent?