Eine skalierbare Web-Datenpipeline aufbauenMit Crawlbase

Eine Web-Datenpipeline ist nur so zuverlässig wie ihre Erfassungsschicht. Wenn Dashboards veralten oder Zahlen nicht mehr stimmen, liegt die Ursache fast nie im Analyse-Code. Sie liegt am Anfang der Pipeline: ein Scraper, der nach einem Website-Redesign kaputt gegangen ist, Anfragen, die anfingen blockiert zu werden, oder Seiten, die sich im Browser laden, aber einer einfachen HTTP-Anfrage eine leere Hülle zurückgeben. Wenn man die Erfassung als fragil behandelt, erbt die gesamte Pipeline diese Fragilität.

Diese Anleitung zeigt Ihnen, wie Sie eine skalierbare Web-Datenpipeline aufbauen, mit Crawlbase für die Erfassung und Standard-ETL-Tooling für den Rest. Sie erfassen Seiten mit der Crawling API für On-Demand-Arbeit und dem asynchronen Crawler für Hochvolumen-Jobs, transformieren und validieren das rohe HTML, laden saubere Zeilen in den Speicher und planen das Ganze mit Monitoring. Jeder Schritt hat ausführbaren Code, den Sie anpassen können.

Wie eine skalierbare Web-Datenpipeline aussieht

Das Muster ist die klassische ETL-Form mit einer wichtigen Trennung von Zuständigkeiten. Crawlbase sitzt vorne als Erfassungsschicht und verarbeitet alles, was Scraping instabil macht: JavaScript-Rendering, IP-Rotation, Anfragen-Routing und Block-Abwehr. Ihre Systeme übernehmen Parsing, Validierung, Speicher und Analyse. Der Ablauf liest sich von links nach rechts:

Web  ->  Crawlbase (collect)  ->  Transform + Validate  ->  Storage  ->  BI / ML

Der Grund für diese Grenzziehung ist Beständigkeit. Externe Websites sind keine stabilen Abhängigkeiten; sie veröffentlichen Layout-Änderungen, führen Experimente durch und setzen Anti-Bot-Abwehr ohne Vorwarnung ein. Indem man eine verwaltete Erfassungsschicht davor setzt, wird eine Website-Änderung zu einer Konfigurationsfrage statt zu einem Pipeline-Ausfall. Crawlbase bietet Ihnen zwei Erfassungstools für zwei Workload-Formen, und eine Produktionspipeline verwendet normalerweise beide.

• Crawling API für Echtzeit-On-Demand-Abruf bekannter URLs. Sie senden eine URL, sie gibt die Seite zurück.
• Async Crawler für großvolumige, Fire-and-Forget-Erfassung. Sie pushen URLs, er ruft sie asynchron ab und POSTet Ergebnisse an Ihren Webhook.

Das ist dieselbe Trennung, zu der jede ernsthafte E-Commerce-Web-Scraping-Operation gelangt: ein schneller Pfad für gezielte Lookups und ein Bulk-Pfad für die Abdeckung. Wenn Sie mit den Proxy-Mechanismen darunter neu sind, ist Was ist ein Proxy-Server nützliches Hintergrundwissen, obwohl der Sinn einer verwalteten API darin besteht, dass Sie nichts davon selbst verwalten müssen.

Schritt 1: Mit der Crawling API erfassen

Die Crawling API nimmt eine URL plus Ihren Token und gibt die gerenderte Seite zurück. Sie senden ein HTTP GET; sie leitet die Anfrage durch einen rotierenden IP-Pool, rendert optional JavaScript, wenn Sie einen JS-Token übergeben, und gibt das HTML (oder geparste JSON) zurück. Der einfachste mögliche Aufruf ist ein einzelnes curl:

curl 'https://api.crawlbase.com/?token=YOUR_TOKEN&url=https%3A%2F%2Fexample.com%2Fproducts'

In einer Pipeline möchten Sie einen kleinen, wiederverwendbaren Collector statt rohem curl. Installieren Sie den offiziellen Client und verpacken Sie den Aufruf, damit der Rest der Pipeline sauberes HTML erhält und nie an Tokens oder Wiederholungen denken muss. Verwenden Sie den JS-Token für clientseitig gerenderte Seiten und den normalen Token für statisches HTML.

python3 -m venv venv && source venv/bin/activate
pip install crawlbase

from crawlbase import CrawlingAPI

api = CrawlingAPI({'token': 'YOUR_TOKEN'})

def collect(url, render=False):
    options = {'ajax_wait': True, 'page_wait': 3000} if render else {}
    response = api.get(url, options)
    status = response['status_code']
    if status != 200:
        raise RuntimeError(f'collect failed for {url}: {status}')
    return response['body'].decode('utf-8')

html = collect('https://example.com/products', render=True)
print(len(html), 'bytes')

Zwei Details machen das pipeline-tauglich statt zum Spielzeug. Erstens prüft es status_code und wirft bei allem, was kein sauberer Abruf ist, eine Ausnahme, sodass eine fehlerhafte Seite laut auftaucht, anstatt Ihr Warehouse mit leeren Zeilen zu vergiften. Zweitens hält das render-Flag Ihre Aufrufstellen ehrlich darüber, welche Seiten JavaScript benötigen: zahlen Sie die Rendering-Kosten nur dort, wo der Inhalt es tatsächlich erfordert. Dieser Collector ist die Einheit, die Ihr Scheduler für jede bekannte URL aufrufen wird.

Schritt 2: Volumen mit dem async Crawler skalieren

Die Crawling API ist synchron: eine Anfrage, eine Antwort, und Ihr Code wartet. Das ist genau richtig für einige hundert bekannte URLs. Bei Zehntausenden skaliert das Blockieren auf jeden Aufruf nicht. Der asynchrone Crawler kehrt das Modell um. Sie pushen URLs in einen benannten Crawler, die Anfrage kehrt sofort mit einer Request-ID zurück, Crawlbase ruft die Seite im Hintergrund ab, und wenn sie fertig ist, POSTet sie das Ergebnis an Ihren Callback-Endpunkt. Nichts in Ihrem Code blockiert beim Warten auf Seiten.

Sie wählen den asynchronen Modus durch Hinzufügen von zwei Parametern zum selben Endpunkt: callback=true und crawler=YourCrawlerName (Sie erstellen den Crawler einmalig im Dashboard und richten ihn auf Ihre Webhook-URL). Eine URL zu pushen sieht so aus:

curl 'https://api.crawlbase.com/?token=YOUR_TOKEN&callback=true&crawler=my-pipeline&url=https%3A%2F%2Fexample.com%2Fp%2F123'

Anstatt des Seitenkörpers erhalten Sie eine Request-ID zurück, was bedeutet, dass die URL in der Warteschlange ist:

{ "rid": "1e92e8bf4618772871c14d4" }

Von Ihrer Seite aus ist das Pushen eines großen Batches eine enge Schleife. Der Punkt ist Durchsatz: Sie feuern alle URLs ab, ohne auf deren Abschluss zu warten, und die Warteschlange absorbiert die Arbeit.

from crawlbase import CrawlingAPI

api = CrawlingAPI({'token': 'YOUR_TOKEN'})

def push_batch(urls):
    options = {'callback': True, 'crawler': 'my-pipeline'}
    for url in urls:
        response = api.get(url, options)
        rid = response['body']['rid']
        print(f'queued {url} -> {rid}')

push_batch([
    'https://example.com/p/123',
    'https://example.com/p/124',
    'https://example.com/p/125',
])

Die andere Hälfte von async ist der Callback-Handler. Crawlbase POSTet die gecrawlte Seite an den Webhook, den Sie beim Crawler registriert haben, und sendet das HTML im Anfragekörper und Metadaten (die Request-ID, Original-URL und Status) in den Headern. Ihr Handler sollte das Minimum tun: schnell mit einem 200 bestätigen und die Nutzlast an Ihren Transform-Schritt übergeben. Schweres Parsing inline zu machen riskiert ein Timeout der Lieferung und eine Wiederholung.

const express = require('express')
const app = express()

// Crawlbase POSTs raw HTML; capture the body as text
app.use(express.text({ type: '*/*', limit: '10mb' }))

app.post('/crawlbase/callback', (req, res) => {
  const rid = req.headers['rid']
  const url = req.headers['url']
  const status = req.headers['original_status']

  // ack immediately, process out of band
  res.sendStatus(200)

  enqueueForTransform({ rid, url, status, html: req.body })
})

app.listen(8080, () => console.log('callback listening on :8080'))

Wenn Sie lieber keinen Webhook betreiben möchten, richten Sie den Crawler auf Crawlbase Cloud Storage und pollieren Sie stattdessen; der Kompromiss ist eine kleine Verzögerung gegenüber null Infrastruktur. In jedem Fall ermöglicht das asynchrone Modell, Millionen von Seiten zu erfassen, ohne dass Ihre Anwendung jemals auf einen Abruf blockiert.

Schritt 3: Das rohe HTML transformieren und validieren

Die Erfassung gibt Ihnen HTML. Der Transform-Schritt verwandelt dieses HTML in saubere, typisierte Datensätze und verwirft alles, was eine Qualitätshürde nicht besteht. Hier rottet eine Menge Pipelines still vor sich hin: ein Job meldet Erfolg, aber die geschriebenen Zeilen sind leer, weil ein Selektor gedriftet ist. Validieren Sie explizit, damit ein Parse-Fehler wie ein Fehler aussieht.

Parsen Sie mit was auch immer zu Ihrem Stack passt; das Beispiel verwendet BeautifulSoup. Die Funktion extrahiert Felder, normalisiert sie in native Typen und weigert sich, einen Datensatz mit fehlendem Namen oder nicht parsbarem Preis auszugeben.

import re
from bs4 import BeautifulSoup

def transform(html, source_url):
    soup = BeautifulSoup(html, 'html.parser')
    records = []

    for card in soup.select('.product-card'):
        name = card.select_one('.title')
        price = card.select_one('.price')
        if not name or not price:
            continue  # skip incomplete cards, do not emit junk

        digits = re.sub(r'[^\d.]', '', price.get_text())
        if not digits:
            continue

        records.append({
            'name': name.get_text(strip=True),
            'price': float(digits),
            'source_url': source_url,
        })

    if not records:
        raise ValueError(f'no records parsed from {source_url} (selectors may have drifted)')

    return records

Die Form, die zählt: Bereinigen Sie jedes Feld in einen nativen Typ (ein Float-Preis, ein getrimmter String), lassen Sie unvollständige Datensätze fallen, anstatt Leerzeichen zu schreiben, und werfen Sie eine Ausnahme, wenn eine ganze Seite nichts ergibt, sodass ein gedrifteter Selektor noch am selben Tag gefunden wird, an dem er kaputt geht, und nicht Wochen später in einem Bericht. Wenn Sie das Parsing für unterstützte Sites komplett überspringen möchten, gibt die Crawling API strukturiertes JSON direkt zurück und dieser Schritt wird zum Durchlauf.

Schritt 4: In den Speicher laden

Mit validierten Datensätzen schreiben Sie sie irgendwo abfragbar hin. Das Ziel hängt von Skalierung und Verwendung ab: eine relationale Datenbank wie PostgreSQL für transaktionalen Zugriff, ein Warehouse wie BigQuery für Analysen, ein Suchspeicher oder eine nachgelagerte Streaming-Plattform. SQLite reicht aus, um das Muster zu zeigen, und das Muster ist das, was verallgemeinert: Upsert auf einem stabilen Schlüssel, sodass ein erneutes Ausführen der Pipeline bestehende Zeilen aktualisiert, anstatt sie zu duplizieren.

import sqlite3
from datetime import datetime, timezone

def load(records, db_path='pipeline.db'):
    conn = sqlite3.connect(db_path)
    conn.execute('''
        CREATE TABLE IF NOT EXISTS products (
            source_url TEXT PRIMARY KEY,
            name TEXT NOT NULL,
            price REAL NOT NULL,
            collected_at TEXT NOT NULL
        )''')

    now = datetime.now(timezone.utc).isoformat()
    for r in records:
        conn.execute('''
            INSERT INTO products (source_url, name, price, collected_at)
            VALUES (?, ?, ?, ?)
            ON CONFLICT(source_url) DO UPDATE SET
                name=excluded.name,
                price=excluded.price,
                collected_at=excluded.collected_at
        ''', (r['source_url'], r['name'], r['price'], now))

    conn.commit()
    conn.close()

Der Upsert macht den Lade-Schritt idempotent: Denselben Batch zweimal auszuführen hinterlässt die Tabelle im selben Zustand, was genau das ist, was Sie wollen, wenn ein Scheduler einen fehlgeschlagenen Lauf wiederholt. Der collected_at-Zeitstempel gibt Ihnen ein Frischeindikator, den Sie im nächsten Schritt für das Monitoring verwenden werden. Ersetzen Sie die SQLite-Aufrufe durch Ihren Warehouse-Client und die Logik überträgt sich unverändert.

Schritt 5: Automatisieren, planen und überwachen

Die Teile fügen sich zu einer Pipeline-Funktion zusammen, und diese Funktion ist das, was Ihr Scheduler aufruft. Das Verdrahten von Collect, Transform und Load mit einem try/except pro URL verhindert, dass eine fehlerhafte Seite einen gesamten Lauf abbricht.

import logging

logging.basicConfig(level=logging.INFO)
log = logging.getLogger('pipeline')

def run_pipeline(urls):
    ok, failed = 0, 0
    for url in urls:
        try:
            html = collect(url, render=True)
            records = transform(html, url)
            load(records)
            ok += 1
        except Exception as err:
            failed += 1
            log.error('pipeline failed for %s: %s', url, err)

    log.info('run complete: %d ok, %d failed', ok, failed)
    if failed > ok:
        raise RuntimeError('majority of URLs failed, check upstream')

Um sie nach einem Zeitplan auszuführen, ist die einfachste Option cron. Dieser Eintrag führt die Pipeline alle sechs Stunden aus und hängt die Ausgabe an ein Log an, das Sie verfolgen oder an Ihren Monitoring-Stack senden können:

# run the pipeline every 6 hours
0 */6 * * * /path/to/venv/bin/python /path/to/run.py >> /var/log/pipeline.log 2>&1

Cron ist für eine Handvoll Jobs in Ordnung. Sobald Sie Abhängigkeiten zwischen Schritten, Wiederholungsversuche und Backfills haben, wechseln Sie zu einem Workflow-Orchestrator wie Apache Airflow oder Prefect, die Ihnen DAGs, automatische Wiederholungsversuche und eine UI für den Laufverlauf bieten. Für den asynchronen Crawler gibt es auf der Erfassungsseite überhaupt keinen Scheduler: Sie pushen URLs und Ergebnisse streamen in Ihren Callback, wenn sie fertig sind.

Monitoring ist der Unterschied zwischen einer Pipeline, der Sie vertrauen, und einer, die Sie beaufsichtigen. Verfolgen Sie mindestens drei Dinge. Volumen: Zeilenzählungen pro Lauf, sodass ein plötzlicher Abfall ein Erfassungsproblem anzeigt. Frische: die collected_at-Zeitstempel, die Sie gespeichert haben, sodass Sie warnen können, wenn Daten veralten. Fehlerrate: das Ok-versus-Fehlgeschlagen-Ergebnis jedes Laufs, sodass ein schleichender Anstieg Sie warnt, dass sich eine Zielseite ändert, bevor alles kaputt geht. Kombinieren Sie das mit vernünftiger Scraping-Hygiene; Wie man Websites scrapt, ohne blockiert zu werden behandelt die Praktiken, die die Erfassungsschicht bei Skalierung gesund halten.

Häufig gestellte Fragen

Wie baue ich mit Crawlbase eine skalierbare Web-Datenpipeline?

Verwenden Sie Crawlbase als Erfassungsschicht und Standard-ETL-Tooling für den Rest. Erfassen Sie Seiten mit der Crawling API für bekannte URLs und dem async Crawler für hohe Volumen, transformieren Sie das zurückgegebene HTML in validierte typisierte Datensätze, laden Sie sie mit einem idempotenten Upsert in den Speicher und planen Sie den Lauf mit cron oder einem Orchestrator, während Sie Volumen, Frische und Fehlerrate überwachen. Crawlbase übernimmt Rendering, IP-Rotation und Block-Abwehr, sodass Ihr Code nur mit sauberen Daten zu tun hat.

Wann sollte ich die Crawling API vs. den async Crawler verwenden?

Verwenden Sie die Crawling API, wenn Sie eine bekannte Liste von URLs haben und die Seite sofort zurück haben möchten, was für Backend-Dienste, Monitoring-Jobs und Echtzeit-Lookups geeignet ist. Verwenden Sie den async Crawler, wenn Sie in hohem Volumen erfassen oder Fire-and-Forget-Lieferung wünschen: Sie pushen URLs, erhalten sofort eine Request-ID und Crawlbase POSTet jedes Ergebnis an Ihren Callback, wenn es fertig ist. Viele Pipelines nutzen beides: die API für gezieltes Retrieval und den Crawler für breite Abdeckung.

Wie funktioniert der async Crawler Callback?

Sie erstellen einen benannten Crawler im Dashboard und richten ihn auf Ihre Webhook-URL, dann pushen Sie URLs mit callback=true und crawler=YourCrawlerName. Jeder Push gibt sofort eine Request-ID zurück. Wenn Crawlbase eine Seite fertig abgerufen hat, sendet es einen HTTP POST an Ihren Webhook mit dem HTML im Körper und Metadaten in den Headern. Ihr Handler sollte schnell einen 200 zurückgeben und die Nutzlast außerhalb des Bandes verarbeiten, damit die Lieferung nicht ausläuft.

Muss ich noch Proxys verwalten oder Anti-Bot-Abwehr handhaben?

Nein. Die Crawling API und der Crawler leiten Anfragen durch einen rotierenden IP-Pool, rendern JavaScript, wenn Sie einen JS-Token übergeben, und wenden Block-Abwehr serverseitig an. Sie senden eine URL und erhalten eine Seite zurück, sodass Sie auf den Betrieb eines eigenen Proxy-Pools und einer Headless-Browser-Flotte verzichten können. Wenn Sie nur rohe rotierende IPs für Ihren eigenen Stack benötigen, stellt der Smart AI Proxy dasselbe Netzwerk als Standard-Proxy-Endpunkt bereit.

Wie verhindere ich, dass meine Pipeline leere oder schlechte Daten schreibt?

Validieren Sie im Transform-Schritt. Prüfen Sie den Antwortstatus bei der Erfassung und werfen Sie eine Ausnahme bei allem, was kein sauberer Abruf ist, dann lassen Sie beim Parsing Datensätze mit fehlenden Pflichtfeldern fallen und werfen Sie eine Ausnahme, wenn eine ganze Seite null Datensätze ergibt, da das in der Regel bedeutet, dass ein Selektor gedriftet ist. Machen Sie den Lade-Schritt idempotent mit einem Upsert, damit Wiederholungsversuche keine Zeilen duplizieren, und speichern Sie einen Erfassungs-Zeitstempel, damit Sie die Frische überwachen und warnen können, wenn Daten veralten.

Kann diese Pipeline Millionen von Seiten verarbeiten?

Ja. Der Engpass in einem naiven Design ist das Blockieren auf jeden synchronen Abruf, was der async Crawler beseitigt, indem er Arbeit in die Warteschlange stellt und Ergebnisse via Callback liefert. Pushen Sie große Batches ohne zu warten, lassen Sie die Warteschlange die Last absorbieren und verarbeiten Sie Ergebnisse, wenn sie eintreffen. Für sehr große oder laufende Programme fügt ein Enterprise-Plan den Durchsatz und Support hinzu, den hochvolumige Erfassung benötigt.

IK

Ian Kalvin

Technical Support Engineer · Crawlbase

Technical Support Engineer bei Crawlbase, der von der Front darüber schreibt, was beim Scraping und bei Proxy-Setups im Produktivbetrieb tatsächlich kaputtgeht.

NZ

Neil Zamora

Senior Architect · Crawlbase

Senior Architect bei Crawlbase mit Fokus auf den Systemen hinter großflächigem Crawling: Proxy-Rotation, Anti-Bot-Resilienz und den APIs, die diese Komplexität verbergen.