A search index with a billion documents returns thousands of candidates matching a query. Showing all of them ranked by TF-IDF doesn’t produce great results. The ranking signals needed for good results (machine-learned models, user personalization, freshness, domain authority) are expensive to compute. You can’t compute them for a thousand candidates in under 100ms. The solution: a funnel. Use cheap signals to eliminate most candidates fast, then apply expensive signals to the small remainder.

A user searches for “couch.” The documents say “sofa.” Without query expansion, no results. With it, the search system understands that “couch” and “sofa” are synonyms and expands the query to match either. This seemingly simple addition improves recall dramatically, but introduces its own set of problems. Synonym Expansion The most basic form: maintain a synonym dictionary. Before executing the query, expand each term with its synonyms. "couch" → ["couch", "sofa", "settee", "loveseat"] Query: "buy couch" becomes: (buy) AND (couch OR sofa OR settee OR loveseat) Synonym dictionaries are domain-specific.

An inverted index maps terms to the list of documents containing them. A web-scale inverted index has billions of documents and trillions of term-document pairs. It doesn’t fit on one machine. You have to partition it. Two Partitioning Strategies Document partitioning: split documents across machines. Machine 1 has documents 1-1M, machine 2 has documents 1M-2M, and so on. Each machine builds a complete local inverted index over its documents. A query for “distributed systems” goes to every machine in parallel, each returns its local top-K results, a merger combines and re-ranks.

Before PageRank, search engines ranked results by how many times the search term appeared on the page. That was gameable: stuff a page with keywords and you’d rank for them. PageRank’s insight was that a link from one page to another is an implicit endorsement. A page linked to by many high-quality pages is probably high quality itself. The Algorithm Each page starts with a rank of 1/N (where N is the total number of pages).

A web crawler visits URLs, extracts content, discovers new URLs from links, and repeats. The loop is simple. Doing it at Google’s scale, visiting hundreds of billions of pages without overwhelming any single website, requires real engineering. We covered priority queues, content fingerprinting, and checkpointing in the context of web crawling. Those are the mechanics of the frontier. This post is about the distributed architecture around it. The Crawl Frontier The frontier is the queue of URLs to visit.

Generating a thumbnail is a one-time cost. Serving it is a continuous cost. For a platform with a billion photos, every profile picture load, every post thumbnail, every image preview is a serve request. Getting the serving architecture right determines whether your image infrastructure is a background cost or a constant fire drill. URL Structure as Cache Key Every image variant needs a stable, unique URL. The URL structure defines the cache key for the CDN.

Two images of the same photo. One was uploaded as a JPEG, the other was screenshotted, scaled 5%, and re-uploaded as a PNG. They look identical to a human. Their SHA-256 hashes share zero bits in common. Cryptographic hashing detects exact duplicates. Perceptual hashing detects near-duplicates. Why Cryptographic Hashes Don’t Work A single pixel difference changes SHA-256 completely. Re-encoding, resizing, or adding a watermark produces a completely different cryptographic hash. Detecting “this is the same photo someone already uploaded” requires a similarity measure, not an equality check.

A user uploads a photo. The platform needs: a thumbnail at 150x150, a medium preview at 800x600, the original compressed to 80% quality, EXIF data extracted, content safety scanning, and a perceptual hash for deduplication. None of these should block the upload response. All of them need to happen reliably. The Fan-Out After Upload The upload endpoint does one thing: store the raw file, acknowledge the upload, publish an event. Everything else is async.

A user uploads a 500MB video. 90% through, their mobile connection drops. They reconnect and have to start over. That’s the worst case for user experience and a waste of everyone’s resources. Resumable uploads solve this by making partial progress durable. Why Standard HTTP Uploads Break A standard HTTP POST sends the entire file as one request body. The server processes it when the full body is received. If the connection drops at byte 490MB of a 500MB upload, the server discards everything received so far.

Changing a config value in etcd is fast. Ensuring every running instance of every service has picked up that change, and knowing with confidence that they all have, is slower and harder. The Propagation Problem A service runs on 200 instances. You update a config key. The etcd watch mechanism notifies all watchers within milliseconds. But watchers are long-lived connections: a service instance that restarted recently might not have its watch re-established yet.

A misconfigured timeout brought down a service. The operator changed one value in a config file. Services that picked up the change started timing out on all requests. To fix it, you need to know what changed, when, and be able to revert it in under a minute. That requires config versioning. Config as Versioned Records Every write to a config store should record: what key changed, what the new value is, what the old value was, who made the change, and when.

Kubernetes stores all its state in etcd. Every pod spec, every deployment, every service endpoint. When a pod is scheduled, etcd records it. When a service endpoint changes, etcd records it. Everything that needs to be consistent across all Kubernetes control plane components lives there. etcd is a strongly consistent key-value store built on Raft. Understanding why you’d want this for configuration, and specifically the watch mechanism that makes it useful, is the interesting part.

Surge pricing reacts to imbalance. Driver repositioning tries to prevent it. If the system knows which zones will be in high demand in the next 30 minutes, it can suggest that idle drivers reposition there before the surge starts. That prediction is the hard part. Demand Forecasting Per Zone Historical patterns are strong for predictable events: the airport zone is high demand every Friday evening. The downtown zone surges at bar close time on weekends.

New Year’s Eve at midnight. Thousands of people try to book rides simultaneously. Drivers are outnumbered 10 to 1. Without any intervention, every rider waits an hour. Surge pricing’s goal is to make more drivers available (higher fares attract drivers on the fence about working) and reduce demand (some riders opt for alternatives). The system design question is how you compute and apply it. The Core Computation Surge is a function of the demand-to-supply ratio in a geographic zone.

“Send the nearest driver” sounds like the right dispatch rule. It minimizes pickup time for this one rider. But it ignores the driver’s position after the drop-off, other pending rider requests, and overall system efficiency. Real dispatch is an optimization problem. Nearest Driver Is a Greedy Heuristic Greedy nearest-driver assignment: find the closest available driver, assign them, done. Fast to compute, easy to understand. The problem is it optimizes for one request in isolation.

Every Uber driver’s phone sends a GPS update every 4-5 seconds. Uber has around 5 million active drivers globally. That’s about 1 million location updates per second hitting the write path. Most systems are read-heavy. Driver tracking is unusually write-heavy, and the data model is unusual too: old locations are worthless, only the current one matters. The Data Model A naive approach: a driver_locations table with columns driver_id, latitude, longitude, updated_at.

A live stream ends. The viewer who missed it wants to watch it on demand. The platform already has all the segments: they were generated during the live stream and pushed to the CDN. Making them into a VOD (video on demand) is less a technical challenge and more a stitching-and-indexing problem. Segments Are Already There During a live stream, transcoding at ingest generates 2-6 second segments at each quality level.

A popular Twitch stream has 200,000 concurrent viewers. Every one of them can send chat messages. A message from one viewer needs to reach all other viewers within a second. That’s fundamentally different from the fan-out problem in a news feed, and the solutions are different too. Why News Feed Fan-Out Doesn’t Apply News feed fan-out is about delivering content to followers: write to each follower’s timeline, or read and merge at request time.

Standard HLS has 15-30 seconds of latency between the streamer’s action and the viewer seeing it. For a live sports match, that’s an eternity. Viewers checking social media see the goal before they see it on stream. Low-latency HLS and related approaches cut this to 2-5 seconds. Where the Latency Comes From Standard HLS segments are 6-10 seconds long. The viewer’s player buffers 2-3 segments before playing. That’s 12-30 seconds of buffering before a single frame plays.

A streamer broadcasts at 1080p60 on a 6Mbps upload. A viewer on a mobile phone with a weak signal can’t play 6Mbps video. A viewer on a 4K TV would love more. The platform needs to serve the same stream at multiple quality levels simultaneously. That’s what transcoding at ingest does. The Ladder A “bitrate ladder” is the set of quality levels a stream is available in. A typical ladder:

Broadcasting live video starts before CDN distribution, before transcoding, before viewers. It starts with getting the video from the streamer’s machine to your servers reliably. That first hop, from broadcaster to origin, is where RTMP fits. RTMP: Designed for Live RTMP (Real-Time Messaging Protocol) was created by Macromedia in the early 2000s for Flash video. It’s old. It’s also still the standard ingest protocol for live streaming because it solves the ingest problem well.

A global leaderboard for a game with 500 million players across 20 countries needs to aggregate scores from everywhere. A player in Mumbai competing for rank 1 globally is competing against a player in São Paulo. The data is split across regional deployments for latency reasons. Combining it without introducing consistency problems is the interesting part. Why Regional Splits Exist Storing all player scores in one region means cross-region write latency for score updates.

A Redis sorted set on a single node handles tens of millions of members with O(log N) operations. That’s enough for most games. Then you have 500 million players and a score update on every match completion. Suddenly one node isn’t enough. The Single-Node Limit A Redis sorted set for 500 million players needs roughly 5-8 GB of memory just for the skip list nodes and hash table entries. More importantly, every score update and rank query hits one CPU core, because Redis is single-threaded per shard.

Redis sorted sets support O(log N) insert, O(log N) rank lookup, and O(log N + M) range queries. A balanced BST like a red-black tree could also do this. Redis uses a skip list instead. For years I just accepted that without understanding why. How a Skip List Works A skip list is a linked list with express lanes. The base level is a sorted linked list of all elements. Above it are increasingly sparse “express lanes”: level 2 has every 2nd element, level 3 every 4th, and so on, each determined probabilistically at insert time.

The namenode knows where data lives. It doesn’t move data. Once a client has chunk locations from the namenode, all data flows directly between the client and datanodes. That separation is what makes HDFS scale. The Write Path Writing a file: The client contacts the namenode, which allocates chunk IDs and assigns 3 datanodes for each chunk based on rack-aware placement. The client sends data to the first datanode. The first datanode simultaneously writes to disk and forwards data to the second datanode.

Three replicas gives you fault tolerance. It also means every byte of storage you buy, you get one-third of it for actual data. For cold storage with petabytes at rest, that cost is hard to justify. Erasure coding is how you get fault tolerance without paying 3x. The Idea Take a chunk of data. Split it into k data fragments. Generate m parity fragments using the data fragments. Store all k+m fragments on different nodes.

Storing 3 replicas of a chunk on 3 different machines sounds like good fault tolerance. It’s not, if all 3 machines are in the same network rack. A rack losing power or a top-of-rack switch failing takes down all 3 replicas simultaneously. You’ve built 3 copies but achieved the fault tolerance of 1. How Racks Fail In a data center, servers are grouped into racks of 20-40 machines. Each rack shares a top-of-rack (TOR) switch and often shares power distribution.

HDFS has a single namenode. Every read and write starts there. For a system designed to scale to petabytes, having one machine that every client must contact sounds like a terrible idea. It kind of is. Understanding why they did it anyway, and how they mitigated it, is the interesting part. What the Namenode Stores The namenode holds all filesystem metadata in memory: the directory tree, file-to-chunk mappings, chunk-to-datanode mappings, and file permissions.

A 10GB video file on a single disk is just a file. A 10GB video file in a distributed system is a problem. How do you store it? Which machine does it go on? What happens when that machine dies? The answer used in GFS and HDFS: split the file into fixed-size chunks, store each chunk on a different machine, and keep the mapping from file to chunks in a separate metadata server.

A route is a sequence of road segments, each with a travel time estimate. Add them up, you have an ETA. That’s the naive version. It’s also wrong often enough that Google spent years making it less wrong. The Simple Version and Its Failures Sum the edge weights on the shortest path. If the router says the trip is 45 minutes, tell the user 45 minutes. This works fine when traffic weights are accurate and conditions are stable.