Orchestrating Distributed Crawls Across Workers #

Problem Framing #

A single-process crawler tops out fast. Beyond a few hundred thousand URLs, one machine cannot hold the frontier in memory, cannot open enough concurrent connections to cover every host at once, and becomes a single point of failure that turns a four-hour audit into an all-night one when it dies at hour three. The fix is not "run a bigger box" — it is splitting the crawl into independent units of work that many machines can process against one shared source of truth. This guide is part of Automated Crawling & Pipeline Tooling and covers the specific mechanics of doing that safely: a shared frontier, consistent-hash sharding by host, a distributed dedup layer, and rate limiting that stays coordinated even though no single process sees every request.

Get this wrong and the failure modes are worse than a slow crawl. Two workers can both fetch the same URL because they never learned the other claimed it first. Five workers can each open three connections to the same rate-limited host and get every one of them banned. A worker that dies mid-shard can silently drop a slice of the site from the audit with no error anywhere. The architecture below exists to make those three failures structurally impossible rather than something you catch in a post-mortem.

---

Prerequisites & Environment Setup #

The reference implementation uses Redis as the shared frontier and coordination layer, Python worker processes, and Docker Compose to run the fleet locally before it goes to a scheduler like Kubernetes or Nomad. Pin every version; a Redis client library mismatch between workers is a common source of "works on one machine, hangs on another" bugs.

Pinned tool versions

Required environment variables

# /etc/crawl-workers/env — sourced by every worker container
export REDIS_URL="redis://frontier.internal:6379/0"
export CRAWL_ID="site-health-2026-07-05"
export NUM_WORKERS="8"
export MAX_RPS_PER_HOST="2.0"
export OUTPUT_BUCKET="gs://site-audit-crawl-artifacts-prod"
export TZ="UTC"

Keep CRAWL_ID unique per run so the frontier, visited set, and output prefix never collide with a previous or concurrent crawl against the same Redis instance.

---

Step 1 — Initialization: Shared Frontier #

Every worker reads from and writes to the same Redis instance rather than holding its own in-memory queue. This is what makes horizontal scaling possible: adding a ninth worker means pointing it at the same REDIS_URL, not redistributing state.

# docker-compose.yml — local fleet for development and load testing
version: "3.9"
services:
  redis:
    image: redis:7.2.4-alpine
    ports:
      - "6379:6379"
    command: ["redis-server", "--appendonly", "yes", "--maxmemory", "2gb", "--maxmemory-policy", "noeviction"]
    volumes:
      - redis-data:/data

  worker:
    build: ./worker
    depends_on:
      - redis
    environment:
      REDIS_URL: "redis://redis:6379/0"
      CRAWL_ID: "site-health-2026-07-05"
      MAX_RPS_PER_HOST: "2.0"
    deploy:
      replicas: 8
    restart: on-failure

volumes:
  redis-data:

# /opt/crawl-workers/frontier.py
# Requires: redis==5.0.4
from __future__ import annotations
import json
import time
from dataclasses import dataclass, asdict

import redis


@dataclass
class FrontierEntry:
    url: str
    depth: int
    host_shard: int


class Frontier:
    """Shared frontier backed by per-shard Redis lists plus a claimed sorted set."""

    def __init__(self, client: redis.Redis, crawl_id: str, visibility_timeout: int = 120):
        self.r = client
        self.crawl_id = crawl_id
        self.visibility_timeout = visibility_timeout

    def _queue_key(self, shard: int) -> str:
        return f"frontier:{self.crawl_id}:shard:{shard}"

    def _claimed_key(self) -> str:
        return f"frontier:{self.crawl_id}:claimed"

    def push(self, entry: FrontierEntry) -> None:
        self.r.rpush(self._queue_key(entry.host_shard), json.dumps(asdict(entry)))

    def claim(self, shard: int) -> FrontierEntry | None:
        """Pop a URL for this shard and record a claim with a visibility timeout."""
        raw = self.r.lpop(self._queue_key(shard))
        if raw is None:
            return None
        entry = FrontierEntry(**json.loads(raw))
        self.r.zadd(self._claimed_key(), {raw: time.time() + self.visibility_timeout})
        return entry

    def ack(self, entry: FrontierEntry) -> None:
        self.r.zrem(self._claimed_key(), json.dumps(asdict(entry)))

    def reclaim_expired(self, shard: int) -> int:
        """Return timed-out claims to their shard queue. Call this from a supervisor loop."""
        now = time.time()
        expired = self.r.zrangebyscore(self._claimed_key(), 0, now)
        for raw in expired:
            entry = FrontierEntry(**json.loads(raw))
            if entry.host_shard == shard:
                self.r.rpush(self._queue_key(shard), raw)
                self.r.zrem(self._claimed_key(), raw)
        return len(expired)

Each shard gets its own Redis list so a worker only ever contends with other workers assigned the same shard, not the entire fleet. The claimed sorted set doubles as an at-least-once delivery guarantee: if a worker dies before calling ack, reclaim_expired returns the URL to circulation after visibility_timeout seconds.

---

Step 2 — Core Configuration: Sharding, Rate Limits, and the Visited Set #

Sharding decides which worker is responsible for which host. Get this parameter wrong and either one worker becomes a bottleneck (too coarse a hash) or per-host politeness collapses because two workers hit the same host at once (sharding by full URL instead of by host).

Key parameters table

// /opt/crawl-workers/shard_config.json (version-controlled)
{
  "crawl_id": "site-health-2026-07-05",
  "num_workers": 8,
  "shard_fn": "consistent_hash_host",
  "max_rps_per_host": 2.0,
  "visited_set": "redis_set",
  "visibility_timeout_s": 120,
  "max_depth": 6
}

# /opt/crawl-workers/sharding.py
# Requires: tldextract==5.1.2
from __future__ import annotations
import hashlib
import tldextract


def registered_domain(url: str) -> str:
    ext = tldextract.extract(url)
    return f"{ext.domain}.{ext.suffix}"


def consistent_hash_host(url: str, num_workers: int) -> int:
    """Assign a whole registered domain to one shard so per-host rate limits
    only ever need to be enforced by a single worker."""
    domain = registered_domain(url)
    digest = hashlib.blake2b(domain.encode(), digest_size=8).digest()
    return int.from_bytes(digest, "big") % num_workers

Enforce max_rps_per_host with a per-host token bucket held in the worker that owns that host's shard — because sharding is by registered domain, exactly one worker ever needs to track that host's bucket, so the rate limit stays coordinated without a distributed lock.

---

Distributed Frontier Architecture #

The diagram below shows how one crawl seed fans out through the shared frontier to sharded workers, each enforcing its own per-host politeness, before results merge back into one artifact.

---

Step 3 — Execution & Graceful Draining #

Launch the worker fleet with docker-compose up --scale worker=8, then monitor per-shard queue depth to catch an imbalanced hash before it stalls the whole crawl. When scaling down or deploying a new worker image, drain in place rather than killing processes outright — an in-flight claim that never gets acknowledged waits out its full visibility_timeout_s before another worker can pick it up.

#!/usr/bin/env bash
# /opt/crawl-workers/run_fleet.sh
set -euo pipefail

CRAWL_ID="${CRAWL_ID:?CRAWL_ID must be set}"
NUM_WORKERS="${NUM_WORKERS:-8}"
LOG="/var/log/crawl-workers/${CRAWL_ID}.log"

mkdir -p "$(dirname "${LOG}")"

{
  echo "=== Distributed crawl started at $(date -u +%Y-%m-%dT%H:%M:%SZ) — ${CRAWL_ID} ==="
  docker-compose up --scale worker="${NUM_WORKERS}" -d

  # Poll per-shard queue depth every 30s until all shards drain to zero
  while true; do
    depths=$(python3 /opt/crawl-workers/queue_depth.py --crawl-id "${CRAWL_ID}")
    echo "queue depths: ${depths}"
    if [[ "${depths}" == *"total=0"* ]]; then
      echo "All shards drained."
      break
    fi
    sleep 30
  done

  echo "=== Distributed crawl completed at $(date -u +%Y-%m-%dT%H:%M:%SZ) ==="
} 2>&1 | tee "${LOG}"

# Graceful drain of a single worker before scale-down or redeploy
docker-compose exec worker-3 kill -SIGTERM 1
# The worker's signal handler stops claiming new URLs, finishes in-flight
# fetches, calls ack() on each, then exits — no reclaim needed on clean drain.

For sizing max_rps_per_host and worker counts against a specific origin's tolerance, see Managing Crawl Budget & Rate Limiting, which covers the token-bucket and backoff mechanics each worker enforces locally. Assignment logic itself is broken out further in Sharding a URL Frontier Across Crawl Workers, and the distributed dedup layer referenced above is covered in depth in Deduplicating URLs in a Distributed Crawl.

---

Step 4 — Artifact Capture: Merging Per-Shard Output #

Each worker writes its own fetch results to a per-shard output file so no two workers contend for the same file handle. A merge step runs once all shards report total=0, reconciling every shard's output against the shared visited set and writing one canonical crawl artifact.

# /opt/crawl-workers/merge.py
# Requires: pandas==2.2.2
from __future__ import annotations
import glob
import os
from datetime import datetime, timezone

import pandas as pd


def merge_shard_outputs(crawl_id: str, shard_glob: str, output_bucket: str) -> str:
    """Concatenate per-shard NDJSON output, drop duplicate URLs, write one Parquet artifact."""
    frames = [pd.read_json(path, lines=True) for path in sorted(glob.glob(shard_glob))]
    if not frames:
        raise RuntimeError(f"No shard output found for crawl_id={crawl_id}")

    merged = pd.concat(frames, ignore_index=True)
    before = len(merged)
    merged = merged.drop_duplicates(subset=["canonical_url"], keep="first")
    after = len(merged)
    print(f"Merged {before} rows -> {after} unique URLs ({before - after} cross-shard duplicates)")

    run_date = datetime.now(tz=timezone.utc).strftime("%Y/%m/%d")
    artifact_path = f"{output_bucket}/crawls/{run_date}/{crawl_id}.parquet"
    merged.to_parquet(artifact_path, compression="snappy", index=False)
    return artifact_path

Cross-shard duplicates should be rare — the visited set exists precisely to prevent them — but the merge step still drops on canonical_url as a defense-in-depth check. A non-zero duplicate count after merge is itself a signal that the dedup layer let something through; treat it as a troubleshooting input, not just cleanup. For long-term artifact handling once the merged file exists, see Storing & Versioning Crawl Artifacts in Cloud Storage.

---

Verification Checklist #

Confirm a distributed run produced a trustworthy artifact before treating it as authoritative:

1. Check the fleet log for All shards drained. — its absence means the run was killed or timed out before completion.
2. Verify per-shard queue depth reached zero on every shard, not just in aggregate: python3 /opt/crawl-workers/queue_depth.py --crawl-id "$CRAWL_ID" --per-shard. A stuck non-zero shard usually means a dead worker whose claims never got reclaimed.
3. Confirm the merged artifact's row count is within expected range of the previous crawl for this domain: python3 -c "import pandas as pd; print(len(pd.read_parquet('path/to/latest.parquet')))".
4. Spot-check the visited-set size against SCARD on the Redis key and compare it to the merged artifact's row count — they should be within a few percent of each other.
5. Confirm no host in the merged output exceeds max_rps_per_host by sampling request timestamps per registered domain; a violation means the per-host token bucket was bypassed, usually by a misconfigured shard_fn.
6. Verify the claimed sorted set (ZCARD on the claimed key) is empty after the run — a nonzero count means URLs are still marked in-flight and were never acknowledged or reclaimed.

---

Troubleshooting #

One worker's queue never drains while the others finish early

Root cause: consistent_hash_host sent a disproportionate number of high-URL-count hosts to the same shard — a hot shard, usually because the fleet is small enough that hash distribution is uneven.

# Compare URL counts assigned per shard
python3 -c "
from sharding import consistent_hash_host
import collections
hosts = open('/tmp/seed_hosts.txt').read().splitlines()
counts = collections.Counter(consistent_hash_host(h, 8) for h in hosts)
print(sorted(counts.items()))
"

Fix: increase num_workers so the hash space subdivides more finely, or move to a weighted assignment that accounts for known high-volume hosts ahead of time rather than relying purely on hash distribution.

---

Two workers both fetched the same URL

Root cause: shard_fn was set to hash the full URL instead of the registered domain, so requests to the same host landed on different shards and no single worker's dedup check could catch the collision before the fetch happened.

# Confirm the config in effect
python3 -c "import json; print(json.load(open('/opt/crawl-workers/shard_config.json'))['shard_fn'])"

Fix: switch shard_fn back to consistent_hash_host (host-level, not URL-level) and re-run — sharding by host guarantees only one worker ever owns a given domain's dedup checks and rate limit.

---

Redis memory grows unbounded mid-crawl

Root cause: the visited set backend is redis_set on a domain with tens of millions of URLs, and each member key consumes real memory with no eviction policy set, or maxmemory-policy was left at a default that evicts frontier data.

redis-cli -u "$REDIS_URL" INFO memory | grep used_memory_human
redis-cli -u "$REDIS_URL" CONFIG GET maxmemory-policy

Fix: confirm maxmemory-policy is noeviction (never silently drop frontier or claimed-set data), and switch visited_set to redis_bloom for large domains — a Bloom filter trades a small false-positive rate for a fixed memory footprint regardless of URL count.

---

A worker restart re-crawls URLs it already fetched

Root cause: the worker held its own in-process dedup cache in addition to checking the shared visited set, and that local cache was lost on restart, causing it to re-issue checks that should have been rejected — or the visited-set write happened after the fetch instead of before the claim.

Fix: write to the visited set immediately after claim() succeeds and before the fetch is issued, not after the fetch completes. This makes the check-then-fetch sequence atomic enough that a mid-fetch crash never causes a duplicate on restart.

---

Crawl finishes but the merged artifact is missing whole sections of the site

Root cause: a worker died and its claimed URLs sat in the claimed set past visibility_timeout_s, but the reclaim loop was never actually running as a scheduled process — it only exists as a function, not a cron job or sidecar.

# Check for entries stuck in the claimed set older than the timeout
redis-cli -u "$REDIS_URL" ZRANGEBYSCORE "frontier:${CRAWL_ID}:claimed" 0 "$(date -d '-5 minutes' +%s)"

Fix: run reclaim_expired on a fixed interval (every 30–60 seconds) as its own supervisor process per shard, not just as a library function called ad hoc — otherwise a dead worker's claims sit unreclaimed indefinitely.

---

Per-host rate limit is respected in aggregate but bursts on individual requests

Root cause: the token bucket refills in large discrete steps (e.g. once per second) rather than continuously, so a burst of requests right after a refill can exceed the intended steady-state rate even though the hourly average looks fine.

Fix: refill the token bucket continuously based on elapsed wall-clock time rather than on a fixed tick, and cap bucket capacity at 1–2x the per-second rate so a refill can never authorize more than a couple of requests at once.

---

Related #

• Automated Crawling & Pipeline Tooling — parent section covering the full crawl-pipeline taxonomy
• Sharding a URL Frontier Across Crawl Workers — consistent-hash assignment in depth
• Deduplicating URLs in a Distributed Crawl — Redis set and Bloom filter dedup patterns
• Managing Crawl Budget & Rate Limiting — the token-bucket and backoff mechanics each worker enforces
• Storing & Versioning Crawl Artifacts in Cloud Storage — what happens to the merged artifact after this run

---

FAQ #

How many workers should a distributed crawl use?

Start from your per-host politeness budget, not from available compute. If a host tolerates 5 requests per second and average response time is 400ms, roughly 2 concurrent connections saturate that host regardless of how many workers you run. Add workers to cover more distinct hosts in parallel, not to hammer one host harder — scaling worker count only helps throughput when the crawl spans many hosts.

What happens if a worker crashes mid-crawl?

A crashed worker leaves its in-flight URLs claimed but unfinished. Use a visibility timeout on the claimed set so an unacknowledged URL is automatically returned to the frontier after a fixed window, and a supervisor process restarts the worker with the same shard assignment. No manual intervention is needed as long as the reclaim loop is actually running as its own scheduled process.

Should sharding be by URL or by host?

Shard by registered domain, not by full URL. Sharding by URL scatters requests to the same host across every worker, making per-host rate limiting impossible to enforce centrally. Sharding by host lets each worker own its own token bucket for every host it is responsible for, so politeness limits never require cross-worker coordination.

Can this architecture crawl multiple sites at once?

Yes. Namespace the frontier, visited set, and output prefix by crawl_id so multiple concurrent site crawls share the same worker fleet and Redis instance without their queues or dedup state colliding. Each crawl's shard keys, claimed set, and merged artifact path are all scoped under its own crawl_id.