Distributed Web Scraping: File Systems and MapReduce

Common Crawl's July 2026 archive holds 2.14 billion pages. Compressed that is 84.69 TiB of WARC, and 364.01 TiB before compression. The crawl ran from 7 to 25 July across 40.5 million hosts and 33.2 million registered domains, which averages out to roughly 1,300 fetched pages a second, sustained for nineteen days.

Now the number that should give you pause. In 2008 a single server crawled 6.3 billion pages in 41 days at an average of 1,789 pages a second. The IRLbot paper (Texas A&M, WWW 2008) ran that on one quad-CPU 2.6 GHz Opteron with 16 GB of RAM and a 24-disk RAID-5 array. One machine, eighteen years ago, sustaining a higher page rate than the largest public crawl in the world averages today.

That comparison is unfair in the ways that matter, and it is still the right place to start. The question is not how to build a distributed crawler. It is what actually forces the work apart, which piece breaks first when it does, and what each piece costs once you own it. Every version, price and limit below was read from vendor documentation, project release pages or package registries on 10 August 2026. Where a figure is ours, the method is written out so you can rerun it.

What actually forces the work apart

The standard answer is throughput, fault tolerance and IP limits. The first is mostly wrong, the second is right for a reason nobody states, and the third is a budget line rather than an architecture.

Throughput is capped by politeness, not by your process. A crawler that respects one request per second per host has a ceiling equal to the number of hosts it may work on concurrently. Common Crawl's 1,300 pages a second is spread over 40.5 million hosts, so the per-host rate is glacial by design. If your target list is one large retailer, five hundred workers do not buy you five hundred times the rate. They buy you five hundred ways to get blocked at once. The earlier version of this article said crawl rate "scales roughly with N", which is the single most misleading sentence a distributed-crawling guide can contain.

State outgrows the box before bandwidth does. The frontier and the seen-set are the parts that stop fitting. Exact deduplication of a billion URLs wants about 76 GiB of RAM, measured further down.

Long runs have to survive being interrupted. A crawl that takes nineteen days will be interrupted. Distribution is mostly a way of making the unit of loss small: one worker dies, its leased URLs go back to the frontier, nothing else notices.

Then the third item, the one that decides the budget. Bandwidth through residential exit addresses is the expensive part of a large crawl, and no architecture changes that. Advertised residential prices on 10 August 2026:

Common Crawl's own ratio gives the multiplier: 84.69 TiB compressed for 2.14 billion pages is about 42 KiB per page, counting request, response and metadata records. Assume wire bytes land in the same range, since servers gzip HTML too. A million pages is then roughly 42 GB, so $105 to $255 of proxy bandwidth per million pages. Storing those same 42 GB costs under a dollar a month. Design accordingly: rotating proxies are the line item, the cluster is rounding error.

Anatomy, with the parts that actually matter marked

        seeds
          │
    ┌─────▼────────────┐        ┌──────────────────────┐
    │  Frontier        │ ◄────► │  Seen-set            │
    │  per-host queues │        │  Bloom / Redis set   │
    └─────┬────────────┘        └──────────────────────┘
          │ leases, not pops
    ┌─────▼──────────────────────────────────┐
    │  Workers (stateless pods or functions) │
    │  DNS → fetch → maybe render → parse    │
    └─────┬────────────────────┬─────────────┘
          │ new URLs           │ batched records
    ┌─────▼──────┐      ┌──────▼──────────────────┐
    │ back to    │      │ Object store            │
    │ frontier   │      │ WARC + Parquet, ~1,000  │
    └────────────┘      │ pages per object        │
                        └──────┬──────────────────┘
                               │
                      ┌────────▼─────────────────┐
                      │ Spark / DuckDB / Athena  │
                      └──────────────────────────┘

Two labels on that diagram are the whole design. Leases, not pops. A worker that pops a URL and dies loses it silently; a worker that leases one with a visibility timeout loses nothing, because the lease expires and the URL comes back. About a thousand pages per object. That number is not aesthetic, and the arithmetic is in the storage section.

The frontier is a scheduler wearing a queue costume

Calling the frontier "a queue" hides its job, which is deciding what may be fetched right now without breaking a host. That means per-host sub-queues, a next-allowed-time per host, priority and depth limits. A single global FIFO gives you none of it.

• Redis is the usual answer, and its licence moved twice while nobody was looking. Up to 7.2 it was BSD-3-Clause, from 7.4 RSALv2 or SSPLv1, and since 8.0 Redis is tri-licensed with AGPLv3 as a third option. Valkey, the Linux Foundation fork created during the closed-licence window, shipped 9.1.1 on 21 July 2026 and remains a drop-in.
• Amazon SQS and RabbitMQ give you leases and redelivery for free, which is the property you wanted.
• Apache Kafka suits crawls that never stop rather than crawls that finish. Current release 4.3.1, 25 June 2026.
• URLFrontier is the one most write-ups miss: a crawler-neutral gRPC API for exactly these operations, Apache 2.0, version 2.5 released 23 October 2025. If you are about to invent a frontier protocol, read theirs first.

Politeness is where the received wisdom is wrong. RFC 9309, which standardised robots.txt in 2022, does not define Crawl-delay at all; it says only that crawlers "MAY interpret other records that are not part of the robots.txt protocol". Google's own documentation is blunter: "other fields such as crawl-delay aren't supported". Common Crawl's FAQ, by contrast, says setting Crawl-delay slows CCBot down. The directive is real, widely honoured by the polite, and not in the standard. RFC 9309 does fix two things worth coding to: parsers must handle at least 500 kibibytes, and a cached robots.txt should not be reused for more than 24 hours.

Scrapy quietly changed sides here. The settings reference now lists CONCURRENT_REQUESTS_PER_DOMAIN as default 1 with a fallback of 8, and DOWNLOAD_DELAY as default 1 with a fallback of 0, because startproject writes the polite value into new projects while the library default stays permissive. Embed Scrapy instead of generating a project and you get 8 concurrent requests per domain and no delay, with nothing to warn you. AutoThrottle is still False by default.

Where the frontier breaks. DNS goes first: forty million hosts means forty million resolutions, and public resolvers rate-limit long before your fetcher does. Robots fetches are the second surprise, one extra request per host per day, which on a host-heavy crawl is a real share of all traffic. Third is skew. A handful of hosts hold most of the URLs, so a naive frontier ends up with millions of queued URLs for three domains and idle capacity everywhere else.

The seen-set, measured rather than described

You will rediscover the same URL constantly, so every crawler needs a fast "seen this?" check. The two options are an exact set of hashes or a Bloom filter, and the usual write-up stops at "Bloom filters are smaller". Here is how much smaller.

We generated 2,000,000 synthetic URLs of the form https://shop{n}.example/a/product/{n}?ref=cat{n}&page={n} and inserted them twice: into a Python set of 8-byte BLAKE2b digests, and into a plain bytearray Bloom filter sized for a 0.1% false-positive rate. Python 3.11.15, memory read from peak RSS.

Measured on a single 2-vCPU container against 2,000,000 URLs, with 1,000,000 never-inserted URLs used to probe the false-positive rate. Your absolute numbers will differ; the 45x gap will not.

Scale it to a billion URLs and the exact set wants about 76 GiB while the filter wants 1.7 GiB. The filter's line is exact, since bits per element is fixed. The set's line is worse than linear, because CPython resizes hash tables in powers of two.

Then read the last column again. A Bloom false positive in a crawler does not produce a wrong answer. It produces a page that is never fetched, with no error, no log line and no way to notice. At a billion URLs and 0.1%, that is a million pages your dataset does not contain. Nobody debugs a missing page they never knew existed. If completeness matters, keep the filter as a cheap front door and confirm hits against exact storage.

Redis has this built in. The Bloom filter type auto-scales rather than failing: "when capacity is reached, an additional sub-filter will be created", at the price of slower lookups as sub-filters stack. Scrapy's own RFPDupeFilter keeps fingerprints in an ordinary in-process set, which is fine for a hundred thousand URLs and is exactly what you replace first.

Workers are the boring part, and should stay boring

A worker leases a URL, resolves it, fetches it, optionally renders it, parses it, emits records and links, and holds nothing. That is the whole contract. Everything interesting lives in the frontier and the store.

Kubernetes is the default host, currently 1.36.2 from 9 June 2026. Serverless is tempting for bursty fetch work and has hard edges: AWS Lambda caps a function at 900 seconds and 10,240 MB of memory, with a default limit of 1,000 concurrent executions. Google also renamed Cloud Functions to Cloud Run functions, so half the tutorials you find describe a product name that no longer exists.

Rendering is the cost multiplier. Playwright 1.62.1 shipped 30 July 2026 and Puppeteer 25.5.0 on 4 August 2026, and either turns a fetch measured in milliseconds into a browser session measured in seconds. Render only what needs it; the dynamic content walkthrough covers how to tell. One wasted second per page across a ten-thousand-page nightly run is close to three hours, and across a million-page crawl it is eleven and a half days of machine time.

Scale also makes you visible. Five hundred workers hitting one origin look nothing like five hundred customers, which is what bot-management platforms are built to notice. Handling that is a separate discipline covering CAPTCHA solving, address rotation and fingerprint coherence, and it is the part of a large crawl most likely to be worth handing to a managed extraction service rather than staffing.

Where the output lives

What GFS and HDFS actually decided

The design decisions are worth knowing precisely, because they get repeated imprecisely. The GFS paper (Ghemawat, Gobioff and Leung, SOSP 2003) states "We have chosen 64 MB, which is much larger than typical file system blocksizes", stores three replicas by default, and keeps "less than 64 bytes of metadata for each 64 MB chunk" so that all metadata fits in the master's memory. HDFS made a different call later: its architecture document says "A typical block size used by HDFS is 128 MB". The honest statement is not "the classic block was 64 to 128 MB", which this article used to say. It is that two systems, seven years apart, picked two different numbers for the same reason.

The replica placement rule is quoted less often and matters more. HDFS puts "one replica on the local machine if the writer is on a datanode, otherwise on a random datanode in the same rack as that of the writer, another replica on a node in a different (remote) rack, and the last on a different node in the same remote rack". One rack can burn without data loss, and the common case still writes locally.

The constraint that kills crawl workloads is metadata, not capacity. Konstantin Shvachko's analysis in USENIX ;login: (April 2010) put it in numbers that have aged well: the NameNode uses "fewer than 200 bytes to store a single metadata object", an average file costs about 600 bytes of heap, and 100 million files therefore need a NameNode with at least 60 GB of RAM. His rule of thumb was 1 GB of metadata per petabyte of storage, and around 10,000 concurrent writers saturate one NameNode.

Now write one HTML file per page. A 500-million-page crawl becomes 500 million file objects plus their blocks, several hundred gigabytes of NameNode heap for maybe 20 TB of actual data. The cluster runs out of metadata long before it runs out of disk. This is the entire reason WARC exists, and the reason Common Crawl packs 2.14 billion pages into files numbering in the hundred thousands: its WET and WAT sets are 100,000 files each, roughly twenty-one thousand records per object.

One correction to the lineage. GFS is not "the original that everything descends from" in any living sense, because Google retired it. Colossus replaced the single master with a horizontally scalable metadata layer stored in BigTable, which Google says let it "scale up by over 100x over the largest GFS clusters", to exabytes across tens of thousands of machines. The single-master design was the part that had to go.

Object storage, and the three things it changed

Most teams do not run HDFS for crawl output in 2026. Hadoop itself is alive, with 3.5.0 released 2 April 2026 carrying 485 fixes over the 3.4 line, but the storage layer under new pipelines is S3, Google Cloud Storage, Azure Blob or Cloudflare R2. Three consequences get missed.

Data locality is over. "Move computation to the data" was the founding principle of the GFS and MapReduce era, and object storage discards it. Compute and storage scale separately and talk over the network on purpose. The phrase survives in articles long after the property did.

Renames are not free and are not atomic. Hadoop's own S3A documentation says "The time to rename a directory is proportional to the number of files underneath it" and "Directory renames are not atomic: they can fail partway through". Commit protocols that stage output and rename it into place are a trap here. Note the scheme too: open-source Spark and Hadoop use s3a://, Apache's original s3:// client "is no longer included in Hadoop", and the s3:// in EMR examples is Amazon's own client. Copy an EMR snippet into a self-hosted Spark job and it fails, without saying why.

Consistency stopped being a problem on 1 December 2020, when S3 became strongly consistent for GET, PUT and LIST at no extra charge. If a tutorial tells you to enable S3Guard or EMRFS Consistent View, it predates the fix.

Two hard limits shape how you write. S3 documents at least 3,500 PUT/COPY/POST/DELETE and 5,500 GET/HEAD per second per partitioned prefix, scaling with more prefixes but "gradually", with 503 Slow Down errors while it catches up. And requests cost money per operation:

Amazon's S3 pricing page builds its tables in the browser and yielded no quotable figure, so no S3 storage price appears above.

Run the write-cost arithmetic. One object per page across a billion-page crawl is a billion Class A operations, which at $0.005 per 1,000 is $5,000 in write requests alone, before a byte of storage. Batch a thousand pages into each WARC and the same crawl costs $5. The small-file penalty on object storage is not a metadata ceiling like HDFS, it is an invoice.

Egress is the other trap, and it is why R2's zero-egress column matters more than its cheaper storage. Keeping July 2026's Common Crawl, 84.69 TiB, costs roughly $1,700 a month on GCS Standard in one region. Downloading it once at $0.12/GB costs about $10,400, which is six months of storage for one copy. Data leaving the cloud it was written in should be a deliberate decision.

If you want a filesystem rather than a bucket, Apache Ozone is the current answer to HDFS's metadata ceiling. It "scales to billions of objects", speaks native S3, and reached 2.2.0 on 15 July 2026.

MapReduce, and what actually replaced it

MapReduce arrived as a 2004 paper by Jeffrey Dean and Sanjay Ghemawat, presented at OSDI'04, whose abstract reports "upwards of one thousand MapReduce jobs are executed on Google's clusters every day". Three stages: map emits key-value pairs, the framework groups values by key, reduce combines each group. Applied to crawl data it counts anything you want, per keyword, per category, per seller, per domain.

Google moved on from writing them by hand almost immediately. FlumeJava (PLDI 2010) exists because "many real-world computations require a pipeline of MapReduces, and programming and managing such pipelines can be difficult", and the Dataflow model paper (VLDB 2015) generalised that into what became Apache Beam. Hadoop still ships MapReduce, described on the project's own front page as "A YARN-based system for parallel processing of large data sets". It works. Almost nobody starts there.

Here is where this article was wrong, and where most articles still are. The claim that Spark is faster because "it keeps intermediate data in memory instead of writing it to disk between stages" is not what Spark does. Spark's own programming guide describes the shuffle plainly: "results from individual map tasks are kept in memory until they can't fit. Then, these are sorted based on the target partition and written to a single file", and it warns that "Shuffle also generates a large number of intermediate files on disk". The shuffle is a disk workload in Spark exactly as it is in MapReduce.

The record-setting run makes the point better than the docs do. When Spark took the large-scale sorting record, Databricks wrote that "using Spark on 206 EC2 machines, we sorted 100 TB of data on disk in 23 minutes", against a previous Hadoop MapReduce record of 2100 machines and 72 minutes, and stated: "All the sorting took place on disk (HDFS), without using Spark's in-memory cache." Three times faster on a tenth of the machines, with the cache switched off. What Spark actually has is a query planner and whole pipelines compiled into one job instead of a chain of separate ones. Memory is a feature, not the explanation.

Spark keeps a fast cadence: 4.2.0 landed 14 July 2026, with 4.1.3 and 4.0.4 a day later and 3.5.9 on 16 July for anyone still on the old line. Spark 4.0 made ANSI SQL mode the default, required JDK 17 and Scala 2.13, and dropped Mesos; the supported cluster managers are now Standalone, YARN and Kubernetes.

Here is the crawl version of word count, written against Spark 4.2.0. The comments mark the two places people lose a day.

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, count, explode, lower, split

spark = (
    SparkSession.builder
    .appName("crawl-analytics")
    # s3a:// is the Hadoop connector. The s3:// in EMR examples is Amazon's own.
    .config("spark.hadoop.fs.s3a.impl", "org.apache.hadoop.fs.s3a.S3AFileSystem")
    .getOrCreate()
)

# Parquet is columnar, so naming the two columns you need keeps the rest of the
# record off the wire. On a crawl table that is most of the bytes.
pages = spark.read.parquet("s3a://crawl-bucket/pages/").select("host", "body")

terms = (
    pages.select("host", explode(split(lower(col("body")), r"\W+")).alias("term"))
         .filter(col("term") != "")
         .groupBy("host", "term")      # this is the shuffle, and it hits local disk
         .agg(count("*").alias("n"))   # this is the reduce
)

# No global sort: orderBy() costs one more shuffle and buys nothing downstream.
# No partitionBy("host") either, because 40 million hosts means 40 million
# directories, which rebuilds the small-file problem by hand.
terms.write.mode("overwrite").parquet("s3a://crawl-bucket/term-counts/")

The column selection is the part that pays. Common Crawl publishes its URL index as Parquet on S3 precisely for this: the columnar index for the July 2026 crawl is 0.20 TiB across 900 files, covering all 2.14 billion pages with columns like url_host_name, fetch_status and warc_filename. Amazon Athena charges $5 per TB scanned, so a full scan of that index costs about a dollar and a two-column query costs cents. Sebastian Nagel's write-up of the layout, published by Common Crawl on 1 March 2018, is still the clearest description of it. For how this layer sits in the wider analytics stack, see big data vs. data mining.

When a cluster is the wrong answer

This is the section the architecture diagrams leave out. Run the numbers before you run Kubernetes.

Ten million pages at Common Crawl's 42 KiB compressed average is about 420 GB. That fits on one NVMe disk with room to spare, and DuckDB 1.5.5, released 22 July 2026, will query it as Parquet on a single machine with no cluster and no scheduler. Polars 1.43.2 does the same in a dataframe idiom. For datasets in the hundreds of gigabytes, one large instance is usually faster end to end than a Spark cluster, because you skip cluster startup, network shuffle and the executor memory settings.

Distribution starts paying when one of these is true, and not before: the working set exceeds one machine's RAM and disk; the crawl must run continuously rather than finish; per-host politeness across tens of thousands of hosts binds before CPU does; or the run has to survive losing a machine without restarting from zero. Everything else is a single box with a good queue.

The crawler side has the same asymmetry. Scrapy 2.17.0, released 7 July 2026 with HTTP/2 and SOCKS proxy support, is maintained by Zyte with more than 500 contributors and will take one machine a long way. The distributed add-ons around it are quieter than the framework. scrapy-redis last shipped 0.9.1 on 6 July 2024, and its requirements.txt still lists six>=1.15, a Python 2 compatibility shim, alongside scrapy>=2.6.0. Frontera, the other classic answer, has not released since v0.8.1 in April 2019. Neither is broken. Neither is moving, and that is worth knowing before you build on top.

The purpose-built distributed crawlers are alive. Apache Nutch 1.22 shipped 20 July 2025 and still runs on Hadoop MapReduce, which is what Common Crawl uses to produce the archive quoted throughout this article. Apache StormCrawler is now a top-level Apache project, at 3.7.0 as of 1 August 2026, built on Apache Storm 3.0.0.

Orchestration and the build path

An orchestrator turns "we ran a crawl" into "we run a crawl". Airflow 3.3.0 arrived 6 July 2026, a year after the 3.0 rewrite; Prefect is at 3.8.2 and Dagster at 1.13.17, both from 7 August 2026. Scheduling is the easy half. The half that matters is idempotency: a stage that reruns after a partial failure has to produce the same output, which means writing to a temporary prefix and swapping, or keying records so duplicates collapse.

A build path with the thresholds attached:

1. One box, async. asyncio and aiohttp, or Scrapy with sensible per-domain limits. Good to a few hundred thousand pages, and further than most people expect. Files on disk, DuckDB for analysis, no queue.
2. One box, external frontier. Move the frontier and seen-set to Redis or Valkey so a restart does not lose the crawl. Still one fetcher, and now the state survives it. This step gets skipped far too often, and it removes most of the pain.
3. Many fetchers, one frontier. Containerised workers leasing from a durable queue, writing batched WARC or Parquet to object storage. Per-host rate limiting has to move to the frontier here, or five hundred polite workers become one rude crawler.
4. Full stack. Kubernetes, Kafka or SQS, Spark, Airflow, managed proxies, and someone whose job is the crawl.

Step two buys the most per unit of effort.

The bottom line

Distributed scraping is two ideas that big data made ordinary: storage spread across machines that can fail, and a processing model that splits work by key. Both are still correct. Most of what surrounds them in older write-ups is not.

What changed: object storage replaced the distributed filesystem and took data locality with it, GFS was retired for Colossus, Spark's speed comes from planning rather than memory, one machine with DuckDB handles datasets that used to need a cluster, and per-host politeness sets your crawl rate instead of worker count. What did not change is metadata pressure, small files, and bandwidth costing more than everything else combined.

If the data matters more than the infrastructure, large-scale crawls delivered as structured datasets skip the whole ladder above. If you are building it yourself, start at step two and measure before you move.