How to Build Distributed Web Scraping Pipelines

· 8 min read ·
How to Build Distributed Web Scraping Pipelines

What You’ll Need

To follow this guide step by step, you will need the following tools and services:

  • n8n Cloud or self-hosted n8n instance for workflow orchestration
  • Hetzner VPS or Contabo VPS for hosting queue infrastructure and distributed workers
  • DigitalOcean as an alternative cloud infrastructure provider
  • Namecheap if you need a custom domain for webhooks and API endpoints

Table of Contents

Architectural Overview of Distributed Web Scraping

When I first started scraping data, single-threaded scripts running locally were more than enough. However, as soon as you need to extract thousands of pages per minute, handle proxy rotation, or bypass memory-heavy Javascript rendering, a single instance crashes under the load. You hit strict IP rate limits, run out of memory, or get stuck waiting on standard network I/O blockages.

Building a distributed web scraping pipeline solves these structural bottlenecks by decoupling task orchestration from task execution. Instead of forcing one script to fetch URLs, parse HTML, handle proxies, and save records, we break the system down into isolated responsibilities:

  1. The Orchestration Layer: Receives triggers, schedules recurring extractions, generates target URLs, and handles failure alerts.
  2. The Message Broker: A central memory store (such as Redis) that manages incoming jobs in a high-throughput queue.
  3. Distributed Worker Nodes: Lightweight instances running Python background consumers (Celery) that process requests concurrently across separate IP addresses.
  4. Data Persistence and Deduplication: A structured storage system like PostgreSQL to guarantee data uniqueness and avoid duplicate page requests.

If you are choosing an engine for orchestrating enterprise workflows, you might want to look at our n8n vs Make vs Zapier: Honest Comparison for 2026 guide. For distributed jobs where full execution control is paramount, combining Python task workers with self-hosted orchestration tools yields the highest flexibility and lowest operational costs.

+------------------+         +-------------------+         +--------------------+
|   n8n Scheduler  |  --->   |    Redis Queue    |  --->   | Celery Worker Node |
|  (Orchestration) |         | (Task Storage)    |         | (Hetzner / DO VPS) |
+------------------+         +-------------------+         +--------------------+
                                                                     |
                                                                     v
                                                           +--------------------+
                                                           | PostgreSQL Database|
                                                           | (Deduplicated Data)|
                                                           +--------------------+

Step 1: Setting Up the Task Queue with Redis and Python Celery

The core engine of our distributed node pool relies on Python’s Celery library coupled with a Redis broker. Celery allows us to define atomic scraping tasks, retry failed network connections automatically, and consume jobs across multiple machines concurrently.

First, let us establish the Python worker code. Save this file as tasks.py:

import os
import requests
from bs4 import BeautifulSoup
from celery import Celery

REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/0")
app = Celery("scraper_tasks", broker=REDIS_URL, backend=REDIS_URL)

app.conf.update(
    task_serializer="json",
    accept_content=["json"],
    result_serializer="json",
    timezone="UTC",
    enable_utc=True,
    task_acks_late=True,
    worker_prefetch_multiplier=1,
)

@app.task(bind=True, max_retries=3, default_retry_delay=10)
def scrape_product_page(self, url: str, proxy_url: str = None):
    proxies = {"http": proxy_url, "https": proxy_url} if proxy_url else None
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
    }
    
    try:
        response = requests.get(url, headers=headers, proxies=proxies, timeout=15)
        response.raise_for_status()
        
        soup = BeautifulSoup(response.text, "html.parser")
        
        title_node = soup.find("h1")
        price_node = soup.find("span", class_="price")
        
        title = title_node.get_text(strip=True) if title_node else "Unknown Title"
        price = price_node.get_text(strip=True) if price_node else "N/A"
        
        return {
            "status": "success",
            "url": url,
            "title": title,
            "price": price,
            "status_code": response.status_code
        }
    except Exception as exc:
        raise self.retry(exc=exc)

In this code, setting task_acks_late=True prevents task loss. If a worker node dies mid-scraping due to network dropped connections or machine failure, Redis will re-assign the task to a remaining healthy worker node. Setting worker_prefetch_multiplier=1 ensures workers do not grab batches of tasks into local memory, keeping job redistribution perfectly balanced across all available server capacity.

If you only need to run single scripts on a fixed schedule without worker queues, check out our guide on Scheduling Python Scripts for Automated Tasks.

💡 Fast-Track Your Project: Don’t want to configure this yourself? I build custom n8n pipelines and bots. Message me with code SYS3-HUGO.

Step 2: Deploying Worker Nodes on Cloud VPS Infrastructure

To execute scrapers across different IP ranges without hitting connection throttling, we distribute our workers across server instances provided by Hetzner VPS or DigitalOcean.

We package our worker environment inside containerized Docker setups. Create a requirements.txt file in your root folder:

celery==5.3.6
redis==5.0.1
requests==2.31.0
beautifulsoup4==4.12.3
psycopg2-binary==2.9.9

Next, write the complete Dockerfile to build your standardized worker image across all hosting providers:

FROM python:3.11-slim

WORKDIR /app

RUN apt-get update && apt-get install -y --no-install-recommends \
    gcc \
    libpq-dev \
    && rm -rf /var/lib/apt/lists/*

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

CMD ["celery", "-A", "tasks", "worker", "--loglevel=info"]

Now create a unified docker-compose.yml to define our core broker host along with two worker processes on the host:

version: '3.8'

services:
  redis:
    image: redis:7-alpine
    container_name: scraper_redis
    restart: always
    ports:
      - "6379:6379"

  worker_node_1:
    build: .
    container_name: worker_node_1
    command: celery -A tasks worker --loglevel=info -c 4
    environment:
      - REDIS_URL=redis://redis:6379/0
    depends_on:
      - redis
    restart: always

  worker_node_2:
    build: .
    container_name: worker_node_2
    command: celery -A tasks worker --loglevel=info -c 4
    environment:
      - REDIS_URL=redis://redis:6379/0
    depends_on:
      - redis
    restart: always

By deploying this same setup on a remote VPS hosted on Hetzner VPS, you can easily scale out horizontally. You simply point remote worker containers back to your central Redis URL using secure SSH tunneling or private VPC networking.

Step 3: Orchestrating Pipelines and Rate Limiting with n8n

While Celery handles the extraction workload, n8n Cloud acts as our control tower. It triggers jobs, listens for webhooks, formats output payload batches, and handles error alerts when proxy services drop connection.

Here is a Python API dispatcher script (producer.py) that n8n can call via HTTP or execute periodically to push targets into the Redis broker queue:

import os
import sys
import json
from tasks import scrape_product_page

def enqueue_target_urls(url_list_file: str):
    if not os.path.exists(url_list_file):
        print(f"Error: File {url_list_file} not found.")
        sys.exit(1)

    with open(url_list_file, "r") as f:
        urls = [line.strip() for line in f if line.strip()]

    tasks_dispatched = 0
    for url in urls:
        scrape_product_page.delay(url=url)
        tasks_dispatched += 1

    print(json.dumps({
        "status": "success",
        "dispatched_count": tasks_dispatched
    }))

if __name__ == "__main__":
    if len(sys.argv) > 1:
        enqueue_target_urls(sys.argv[1])
    else:
        print(json.dumps({"error": "No input file provided"}))

In n8n, you configure an Execute Command node or an HTTP Request node to invoke this producer script when new target catalogs are discovered.

{
  "nodes": [
    {
      "parameters": {
        "rule": {
          "interval": [
            {
              "field": "hours",
              "minutesInterval": 6
            }
          ]
        }
      },
      "id": "1",
      "name": "Schedule Trigger",
      "type": "n8n-nodes-base.scheduleTrigger",
      "typeVersion": 1.1,
      "position": [240, 300]
    },
    {
      "parameters": {
        "command": "python3 /app/producer.py /app/urls.txt"
      },
      "id": "2",
      "name": "Execute Producer",
      "type": "n8n-nodes-base.executeCommand",
      "typeVersion": 1,
      "position": [460, 300]
    }
  ],
  "connections": {
    "Schedule Trigger": {
      "main": [
        [
          {
            "node": "Execute Producer",
            "type": "main",
            "index": 0
          }
        ]
      ]
    }
  }
}

If any critical job fails across our nodes, we can route instant notification payloads directly through dedicated alert pipelines. Just like we configured error logging alerts when building interactive messaging bots in How to Build a Telegram Bot with n8n (No Code Required), you can route scraping failure updates to Telegram or Slack to instantly notify your team of broken DOM selectors or IP blocks.

Step 4: Storage, Deduplication, and Result Aggregation

A distributed scraper running across multiple nodes will inevitably hit redundant URLs if target sites link back to parent categories. To enforce atomic persistence and avoid writing duplicate data, we implement a database layer using PostgreSQL with unique constraints.

Here is the database handler setup (db_writer.py) that parses and safely inserts incoming worker payloads:

import os
import psycopg2
from psycopg2.extras import execute_values

DB_HOST = os.getenv("DB_HOST", "localhost")
DB_NAME = os.getenv("DB_NAME", "scraper_db")
DB_USER = os.getenv("DB_USER", "postgres")
DB_PASS = os.getenv("DB_PASS", "secretpassword")

def init_database():
    conn = psycopg2.connect(
        host=DB_HOST,
        database=DB_NAME,
        user=DB_USER,
        password=DB_PASS
    )
    cur = conn.cursor()
    cur.execute("""
        CREATE TABLE IF NOT EXISTS scraped_records (
            id SERIAL PRIMARY KEY,
            url VARCHAR(2048) UNIQUE NOT NULL,
            title TEXT,
            price VARCHAR(100),
            status_code INT,
            scraped_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
        );
    """)
    conn.commit()
    cur.close()
    conn.close()

def save_scraped_data(url: str, title: str, price: str, status_code: int):
    conn = psycopg2.connect(
        host=DB_HOST,
        database=DB_NAME,
        user=DB_USER,
        password=DB_PASS
    )
    cur = conn.cursor()
    query = """
        INSERT INTO scraped_records (url, title, price, status_code)
        VALUES (%s, %s, %s, %s)
        ON CONFLICT (url) 
        DO UPDATE SET 
            title = EXCLUDED.title,
            price = EXCLUDED.price,
            status_code = EXCLUDED.status_code,
            scraped_at = CURRENT_TIMESTAMP;
    """
    cur.execute(query, (url, title, price, status_code))
    conn.commit()
    cur.close()
    conn.close()

if __name__ == "__main__":
    init_database()
    print("Database initialized successfully.")

By leveraging ON CONFLICT (url) DO UPDATE, our workers remain idempotent. It does not matter if five parallel nodes scrape the exact same URL at the same moment. PostgreSQL will deduplicate incoming streams seamlessly without throwing primary key exceptions or creating dirty data states.

To connect this persistence script to your task runner, update your worker tasks.py to import and call save_scraped_data inside the success block before returning the structured object:

from db_writer import save_scraped_data

# Inside scrape_product_page task after parsing title and price:
save_scraped_data(url=url, title=title, price=price, status_code=response.status_code)

Getting Started

Building resilient, fault-tolerant distributed scrapers allows you to extract millions of records reliably without running into computational bottlenecks or IP bans. By separating orchestrations, task brokers, workers, and atomic relational storage, you can easily scale execution by launching additional nodes as your data extraction requirements grow.

To start building your own scalable infrastructure:

Outsource Your Automation

Don’t have time? I build production n8n workflows, WhatsApp bots, and fully automated YouTube Shorts pipelines. Hire me on Fiverr, mention SYS3-HUGO for priority. Or DM at chasebot.online.

Want to automate this yourself?

Start with n8n Cloud (free tier available) or self-host on a Hetzner VPS for full control.

Want this engine running on your own VPS?

This blog publishes itself — daily, unattended, on free API tiers. The full engine, Hugo theme, and setup guide are available as System 3.

Get System 3

Related Guides

system online