Distributed Web Crawling

When You Need Distribution

A single well-optimized machine with async I/O can typically crawl 500-2000 pages per second, processing roughly 50-170 million pages per day. This is sufficient for many use cases: SEO audits of large sites, competitive monitoring of dozens of retailers, or building datasets of a few million pages. You need distributed crawling when your requirements exceed what one machine can deliver in terms of throughput, network bandwidth, memory for the URL frontier, or geographic diversity for accessing region-locked content.

The most common triggers for going distributed are: needing to crawl more than 100 million pages in a single pass, maintaining freshness across millions of domains where per-domain politeness delays limit single-machine throughput, requiring multiple IP addresses to avoid triggering rate limits, or needing geographic distribution to access content that varies by region. If your crawl fits comfortably on one machine, avoid the complexity of distribution. The operational overhead of coordinating multiple nodes, handling partial failures, and maintaining consistency is substantial.

URL Partitioning Strategies

The fundamental decision in distributed crawler architecture is how to partition the URL space across nodes. The partitioning strategy determines which node is responsible for which URLs, directly affecting politeness enforcement, load balance, and fault tolerance.

Domain-Based Partitioning

The most common approach assigns each domain to exactly one crawler node using consistent hashing of the domain name. All requests to example.com go through the same node, regardless of which node originally discovered the URL. This guarantees correct politeness enforcement: since one node handles all requests to a domain, it can enforce per-domain rate limits without coordination. It also enables efficient connection pooling (one pool per domain) and robots.txt caching (one cache per domain).

The downside of domain-based partitioning is potential load imbalance. Some domains have millions of pages while others have a handful. A node assigned to Amazon.com has vastly more work than one assigned to small personal blogs. Consistent hashing with virtual nodes (assigning each physical node multiple positions on the hash ring) helps distribute load more evenly, but hotspots remain possible. Dynamic rebalancing based on observed load per node provides a further mitigation.

URL-Based Partitioning

An alternative approach hashes individual URLs rather than domains, distributing pages from the same domain across multiple nodes. This achieves better load balance since large domains spread naturally across all nodes. However, it makes politeness enforcement much harder: multiple nodes might simultaneously request pages from the same domain, requiring distributed rate limiting via a shared coordination service. This added complexity makes URL-based partitioning less common despite its load-balancing advantages.

Hybrid Approaches

Some systems use hybrid partitioning: domain-based assignment for most domains, with large domains (those exceeding a threshold number of known pages) further subdivided by URL path hash across multiple nodes. This captures the politeness benefits of domain assignment for the majority of domains while preventing extreme hotspots from the largest sites. The coordination overhead only applies to the small number of domains that exceed the subdivision threshold.

Coordination and Communication

Distributed crawlers need mechanisms for nodes to share discovered URLs, coordinate scheduling decisions, and exchange state information. The choice of coordination infrastructure significantly impacts system performance and operational complexity.

Message Queues for URL Distribution

Apache Kafka is the most popular backbone for distributed crawler coordination. When a node discovers URLs belonging to domains assigned to other nodes, it publishes those URLs to a Kafka topic partitioned by domain hash. Each node consumes from the partitions corresponding to its assigned domains. Kafka's durability guarantees that URLs are not lost if a node crashes before processing them, and its high-throughput design handles the millions of URLs that a large crawl generates per minute.

Redis provides a lighter-weight alternative for smaller distributed crawlers. Redis lists or streams serve as per-node URL queues, with discovered URLs pushed to the appropriate node's queue based on domain hash. Redis is simpler to operate than Kafka but provides weaker durability guarantees (depending on persistence configuration) and lower throughput at extreme scale. The Scrapy-Redis library implements this pattern directly.

Shared State Management

Distributed crawlers need shared state for URL deduplication (has this URL been seen by any node?), crawl status tracking (which pages have been successfully crawled?), and global statistics (total pages crawled, error rates, throughput). Redis is commonly used for deduplication via its SET data type or bloom filter modules, providing O(1) membership testing across all nodes. For very large crawl states (billions of URLs), distributed key-value stores like Apache Cassandra or a sharded Redis cluster handle the capacity.

Service Discovery and Health

Nodes need to know which other nodes are alive and which domain assignments are current. Systems like Apache ZooKeeper, etcd, or Consul provide distributed coordination services for leader election (choosing which node manages rebalancing), configuration distribution (broadcasting partition assignments), and health monitoring (detecting failed nodes). When a node fails, the coordinator reassigns its domains to surviving nodes and replays any unprocessed URLs from the durable queue.

Fault Tolerance

In a distributed system running continuously, node failures are not exceptions but expected events. Hardware fails, networks partition, processes crash from memory leaks or bugs, and deployments require rolling restarts. A distributed crawler must handle all of these without losing data or stalling the crawl.

The key design principles for fault tolerance are: durable queues (URLs in transit survive node failures), idempotent processing (re-processing a URL produces the same result without side effects), checkpointing (each node periodically persists its local state), and timeout-based failure detection (if a node stops sending heartbeats, reassign its work after a grace period). Together, these ensure that any single node can fail and the system recovers automatically by redistributing the failed node's workload across survivors.

Data loss is the most critical failure mode to prevent. URLs discovered but not yet published to the durable queue are lost if the discovering node crashes. Minimizing this window (publishing discovered URLs immediately rather than batching) reduces potential loss. For the storage layer, replicated databases or object stores with durability guarantees ensure that crawled content survives individual storage node failures.

Storage at Scale

Distributed crawlers generate enormous data volumes. At 100KB average per page (including HTML and metadata), crawling one billion pages produces roughly 100TB of raw data. This demands distributed storage systems designed for high write throughput and efficient retrieval.

For raw content storage, cloud object stores (S3, GCS) or distributed filesystems (HDFS) provide effectively unlimited capacity with high durability. Content is typically stored in WARC format for archival use or as compressed files organized by crawl batch and domain. Object stores handle the throughput requirements well since each node writes independently without coordination.

For metadata and crawl state (URL records, page metadata, link graphs), distributed databases like Apache Cassandra, ScyllaDB, or managed services like DynamoDB provide the combination of high write throughput, horizontal scalability, and query flexibility needed for crawl databases. Schema design should optimize for the primary access patterns: writing new records during crawling and reading records for scheduling and deduplication during subsequent crawl rounds.

Real-World Architectures

The Common Crawl project provides a public reference architecture for web-scale distributed crawling. Their system crawls approximately 3 billion pages per monthly crawl, storing results as WARC files in Amazon S3. The crawl uses Apache Nutch running on a cluster of machines with domain-based URL partitioning and Hadoop for batch processing of crawl rounds.

Scrapy-based distributed crawlers represent the most accessible path for teams building their own distributed system. Scrapy-Redis replaces Scrapy's in-memory queue and deduplication filter with Redis-backed equivalents, allowing multiple Scrapy instances to share a frontier and avoid duplicate work. Each instance runs a standard Scrapy spider but pulls requests from and pushes discovered URLs to a shared Redis server. This approach scales to 5-10 nodes effectively and handles tens of millions of pages.

For larger scale (hundreds of nodes, billions of pages), custom architectures using Kafka for URL distribution, Cassandra for crawl state, and S3 for content storage provide the performance and operational characteristics needed. These systems require dedicated infrastructure engineering but achieve throughput measured in tens of thousands of pages per second across the cluster.

Operational Considerations

Running a distributed crawler in production requires monitoring, alerting, and operational tooling beyond the crawler code itself. Essential metrics include per-node throughput (pages per second), error rates by type (timeouts, DNS failures, HTTP errors, parse errors), frontier depth (how many URLs are waiting), storage growth rate, and inter-node communication latency. Anomalies in these metrics indicate problems: a sudden drop in one node's throughput suggests it is overwhelmed or experiencing network issues, while a growing frontier with stable throughput suggests the crawl is discovering pages faster than it can process them.

Capacity planning requires estimating the total crawl size, desired completion time, per-page resource consumption, and overhead for coordination. A useful heuristic: plan for 500 pages per second per node with async HTTP, then add 20-30% overhead for coordination, retries, and uneven load distribution. If you need to crawl 100 million pages in 24 hours, that requires roughly 1,150 pages per second sustained, suggesting 3-4 crawler nodes plus coordination infrastructure.