Design a distributed web crawler

Source: https://www.hack2hire.com/question-bank/questions/system-design/69dc41c3e4441834b3f736bb
Company: Databricks, Microsoft AI, Meta, Microsoft, OpenAI
Difficulty: Medium

Problem statement

Crawl billions of web pages politely and repeatedly, deduplicating URLs and content, to feed a search index or a training-data pipeline.

Requirements

We are building a distributed web crawler that politely fetches billions of pages, re-fetches them as they change, drops duplicate URLs and duplicate content, and streams clean pages to a search indexer or a training-data pipeline. Questions worth settling first: HTML only or also PDFs and images? Do we render JavaScript? Is freshness (news) or coverage (training corpus) the goal? Do we have a seed list and a host blocklist?

Functional requirements

• Start from seed URLs and sitemaps, fetch pages, extract links, and add new links to the crawl.
• Obey robots.txt (allow/disallow rules, Crawl-delay) and a per-host rate limit.
• Fetch each URL at most once per crawl cycle; skip exact and near-duplicate content.
• Re-crawl known pages on a schedule that follows how often each page changes.
• Store raw responses and publish parsed pages (URL, text, links, metadata) to downstream consumers.
• Operators can add seeds, block hosts, and change per-host rate limits without a deploy.

Non-functional requirements

• Throughput: 1B fetches per day, about 12k pages/s on average and 25k/s at peak.
• Politeness is a hard rule: never more than one open connection per host by default, never faster than the host's limit.
• Fault tolerant: a crashed node loses at most a few minutes of in-flight work, and no URL is lost from the frontier.
• Scales horizontally by adding nodes; hosts move between nodes without a full reshuffle.
• Robust against hostile or broken sites: traps, endless redirects, huge pages, slow servers.
• Eventual consistency is fine; a URL fetched twice by mistake costs bandwidth, not correctness.

Back-of-the-envelope estimation

• Known URL space: 20B URLs across about 200M hosts. URL metadata at about 250 B/row → 5 TB, before replication.
• Fetch volume: 1B pages/day × 80 KB average HTML ≈ 80 TB/day raw, about 20 TB/day after gzip. Over 30 days of retention that is 600 TB in object storage.
• Bandwidth: 12k pages/s × 80 KB ≈ 1 GB/s ≈ 8 Gbps average, plan for 20 Gbps at peak.
• Concurrency: average fetch latency is about 1 s (DNS + TCP + TLS + transfer), so we need about 25k requests in flight at peak. At 2,000 open sockets per async fetcher node that is 13 nodes; run 40 so that parsing CPU, slow hosts and failures leave headroom.
• Politeness: at 1 request/s per host, 25k pages/s needs at least 25k hosts active at once. The scheduler must spread work across hosts, not drain one host at a time.
• Link discovery: about 50 outlinks per page → 600k new links/s to deduplicate, most of them already seen.

API design

The parts talk over internal RPC (gRPC) and a log (Kafka). The fetch path is pull-based: fetchers lease work and acknowledge it, so a dead fetcher just lets its leases expire.

Frontier queue contract (one frontier shard owns a set of hosts):

Enqueue(urls: [{url, url_fp, host, depth, priority, source}]) → {accepted: int}
Lease(worker_id, max: 500) → [{lease_id, url, host, ip, etag, last_modified, deadline}]
Complete(lease_id, {http_status, content_hash, simhash, bytes, fetch_ms, changed: bool})
Fail(lease_id, {error: "timeout" | "dns" | "5xx" | "robots_denied", retry_after_s})

Lease returns only URLs whose host is ready (next_allowed_at <= now, no open lease on that host). A lease lasts 5 minutes. If it expires without Complete or Fail, the URL returns to its host queue.

Metadata store interface:

SeenBatch(url_fps: [uint64]) → [bool]           -- for link dedupe
GetUrl(url_fp) → UrlRecord
UpsertFetch(url_fp, fetch_result, next_fetch_at) -- idempotent on (url_fp, fetched_at)
GetHost(host) → HostRecord; UpdateHost(host, patch)

Supporting interfaces: Resolve(host) → {ips, ttl} on each node's caching resolver; Robots(host) → {rules, crawl_delay, fetched_at}; ContentSeen(sha256) → {first_url_fp}; NearDup(simhash) → [url_fp].

Operator and downstream API:

• POST /v1/seeds {urls: [...], priority} and PUT /v1/hosts/{host} {rate_limit_rps, blocked}.
• Kafka topic pages.parsed, keyed by url_fp: {url, final_url, fetched_at, http_status, content_hash, text_ref, links, lang}. Consumers read at their own pace.

Recovery contract: every frontier shard writes its queues to local RocksDB and replicates to one peer. Fetch results are at-least-once. All writes are keyed by url_fp (plus fetched_at for history), so replays do no harm. After a crash, the replacement shard reloads its queues, expires every open lease, and continues.

Data model

URL table: wide-column store (Bigtable, HBase, or Cassandra). It handles 20B rows, heavy random writes, and range scans by host. Row key reversed_host/path (e.g. com.example.www/blog/a) keeps one host's URLs contiguous for per-host scans and re-crawl sweeps.

Host table: same store, keyed by host, about 200M rows: ip, robots_rules, robots_fetched_at, crawl_delay_s, rate_limit_rps, next_allowed_at, consecutive_errors, backoff_until, daily_budget, pages_today, blocked.

Content: raw responses go to object storage (S3/GCS) as WARC files of about 1 GB, path crawl/{date}/{shard}/{seq}.warc.gz. A content index sha256 → {warc_path, offset, first_url_fp} lives in a key-value store so exact duplicates are not stored twice.

Frontier state: local RocksDB on each frontier shard: per-host FIFO queues, a priority index, and a min-heap of (next_allowed_at, host). This is not a shared database, because 200M per-host queues do not map to Kafka topics or Redis lists.

Core invariants:

1. At most one in-flight request per host (per IP for shared hosting), and none before next_allowed_at.
2. A URL disallowed by the host's current robots.txt is never fetched.
3. A url_fp is in the frontier at most once at a time.
4. A host belongs to exactly one frontier shard at a time, so its politeness state has a single writer.

High-level design

Why this is hard: the work is mostly network waiting, the politeness rules force us to spread load across millions of hosts, the URL space grows faster than we can fetch, and the web contains traps that generate endless URLs.

flowchart LR
  OPS[Seeds and sitemaps] --> K1[Kafka discovered urls]
  SCH[Recrawl scheduler] --> K1
  K1 --> DD[URL dedupe]
  DD --> FR[Frontier shards by host]
  FR -->|lease| F[Fetchers async]
  F --> DNS[Caching DNS]
  F --> RB[Robots cache]
  F --> WEB[The web]
  F --> OS[(Object store WARC)]
  F --> P[Parsers]
  P --> CD[Content dedupe]
  CD --> K2[Kafka pages parsed]
  K2 --> IDX[Indexer or training pipeline]
  P -->|outlinks| K1
  P --> MD[(URL and host tables)]
  DD --> MD
  SCH --> MD

1. Crawl loop. A fetcher calls Lease on its frontier shard and gets up to 500 URLs from different hosts that are ready. For each URL it resolves DNS from the local cache, checks the cached robots rules, and sends a conditional GET (If-None-Match, If-Modified-Since). Limits: 10 s timeout, 5 redirects, 10 MB maximum body. It writes the raw response to the WARC writer, passes the body to a parser, and calls Complete. The frontier then sets next_allowed_at = now + max(crawl_delay, 1/rate_limit, 10 × fetch_time), which slows us down on hosts that respond slowly.
2. Content handling. The parser decodes the page, extracts text and links, and computes SHA-256 and a 64-bit SimHash. An exact duplicate stores only a pointer. A near-duplicate is marked so the index can collapse it. Unique pages go to pages.parsed. A 304 Not Modified only updates the URL row.
3. Link discovery. The parser normalizes each outlink: lowercase the scheme and host, drop the fragment and the default port, sort query parameters, strip known tracking parameters, and resolve against <base>. It drops links to blocked hosts, links that fail trap rules, and links past the depth limit. The rest go to Kafka urls.discovered, partitioned by hash(host).
4. Frontier expansion. URL dedupe consumers check each url_fp against a Bloom filter, then confirm a "maybe seen" in the URL table. New URLs get a row and an Enqueue call on the shard that owns the host. Known URLs only add to their in-link count, which raises their priority.
5. Re-crawl. The scheduler scans host ranges of the URL table for next_fetch_at <= now + 1h and puts those URLs back into urls.discovered at their stored priority.

Deep dives

Frontier design: priority and politeness together. Each shard uses two layers of queues. The front layer has about 10 priority queues; a URL's priority comes from in-links, host quality, sitemap hints and re-crawl urgency. The back layer has one FIFO per active host plus a min-heap keyed by next_allowed_at. A selector moves URLs from front to back queues, choosing a front queue at random with weights by priority (for example 40% from the top tier), so low-priority URLs still progress. Lease pops ready hosts from the heap, takes one URL from each, and holds a lease on the host. Only about 1M hosts per shard have a back queue in RAM; the rest wait on disk. Per-host budgets (for example 50k pages/day unless a host is allowlisted) stop one large site from using all the capacity. For IPs that serve many hosts, add a second limit per IP.

I/O concurrency: keeping fetchers busy. A thread per request wastes memory at 2,000+ sockets per node, and a blocking DNS lookup or slow TLS handshake stalls the thread. Use an event loop (epoll with async Rust, Go, or Python asyncio and uvloop) and keep a fixed pool of in-flight requests per node. Keep CPU work out of the loop: decompression, HTML parsing and hashing go to a worker pool on other cores. Reuse connections (keep-alive and HTTP/2) for consecutive URLs on the same host. Set a time limit on each phase (connect 3 s, first byte 5 s, total 10 s) and a minimum transfer speed, so a slow server cannot hold a socket for minutes. One node then does 500–1,000 pages/s, and parsing CPU, not the network, becomes the limit.

URL and content deduplication at scale.

A Bloom false positive means a new URL is skipped by mistake. At 1% this is acceptable for a search crawl. For full coverage, confirm every "maybe seen" in the URL table: about 90% of discovered links are already known, so that adds about 500k reads/s, spread across the table. Batch the lookups (SeenBatch with 1,000 keys) and cache recently seen fps in an LRU per consumer.

Scaling from one machine to a fleet. Partition by host, never by URL: shard = consistent_hash(host) over virtual nodes. Then all of one host's politeness state, robots cache and queue live on one node (invariant 4), and no distributed lock is needed. Kafka urls.discovered is partitioned on the same key, so links to a host flow to the shard that owns it. When a node joins, it takes some virtual nodes. The previous owner hands over each host's queue and host state, and keeps no lease on those hosts after the handover. Very large hosts (Wikipedia, big news sites) can have a negotiated higher rate; they still have a single owner, which uses parallel connections up to the agreed limit. Coordination (shard map, node health) lives in etcd or ZooKeeper. Fetchers in several regions should be near the hosts they crawl, so assign shards to regions by the host's IP location.

Freshness, traps, and DNS.

• Re-crawl: estimate a change rate per URL. After each fetch, halve the interval if the content hash changed and multiply it by 1.5 if it did not, bounded to [1 h, 60 d]. Start news-like and sitemap lastmod URLs at short intervals. Conditional GETs make an unchanged fetch cheap (a 304 is about 300 bytes). Give re-crawls a fixed share of capacity (for example 60%) so discovery does not use it all.
• Traps: limit URL length (2,048), path depth (16 segments), and repeated path segments (/a/b/a/b/). Allow at most 3 hops away from pages with in-links from other hosts. Per-host budgets bound the damage. Calendar pages, session IDs in URLs, and faceted search create endless parameter combinations: when a host produces many distinct URLs whose content hashes collide, shrink its budget and drop the parameters that do not change the content. Detect soft 404s by fetching a random path once per host and comparing SimHash.
• DNS: at 12k pages/s, an uncached lookup (20–200 ms) is a bottleneck, and a public resolver will throttle us. Run a local caching resolver (Unbound) on each node with a minimum TTL of 5 minutes. Resolve hosts in advance when they get a back queue. Cache negative answers for 1 hour. Store the IP in the host row so per-IP politeness works.
• Robots: fetch robots.txt before the first URL on a host and cache it for 24 h. A 4xx response means everything is allowed. A 5xx or timeout means no fetches; retry with backoff.

Trade-offs

• Partitioning by host makes politeness a local, lock-free check, but an outsized host is limited to one shard's capacity. We accept that because politeness limits it anyway.
• A Bloom filter alone uses little memory and is fast, but it silently drops about 1% of new URLs. Confirming in the URL table recovers them at the cost of about 500k extra reads/s.
• At-least-once fetching with leases is simple to recover, but a crash can cause a duplicate fetch. That is acceptable for a web server, but it would not be for a non-idempotent endpoint (we only send GET).
• Adaptive re-crawl intervals spend bandwidth where the content changes, but a page that changes rarely and then suddenly will be stale for up to its current interval.
• Skipping JavaScript rendering keeps fetches cheap (about 1 s and little CPU). Rendering a page in headless Chrome costs 10–50× more, so render only for a list of hosts known to need it.
• Storing raw WARC files costs 20 TB/day, but it lets the indexer or the training pipeline reprocess without fetching again.
• Weighted random choice between priority queues keeps every tier moving but delays the best URLs a little, compared with strict priority.

Interviewer expectations

Mid-level (L4)

• Draws the basic loop: frontier, fetcher, parser, link extraction, storage.
• Mentions robots.txt, a per-host delay, and a visited-URL set.
• Gives rough numbers for pages/s and storage.

Senior (L5)

• Separates priority and politeness in the frontier and partitions by host so politeness needs no locks.
• Uses async I/O, a DNS cache, and conditional GETs, with numbers for concurrency.
• Deduplicates URLs (normalization, fingerprint, Bloom filter) and content (hash and SimHash), and handles traps and re-crawl.

Staff (L6+)

• Defines invariants and a recovery contract (leases, idempotent writes, shard handover) and explains where each one holds.
• Balances discovery against freshness, and explains the change-rate model and capacity split.
• Fits the design to its consumer: freshness and ranking for search, corpus-level MinHash dedupe, licensing and opt-out filters for training data. Also covers multi-region placement and cost.