Construire un pipeline de données web évolutifAvec Crawlbase

Un pipeline de données web n'est aussi fiable que sa couche de collecte. Quand les tableaux de bord se figent ou que les chiffres cessent de s'additionner, la cause n'est presque jamais le code d'analyse. C'est le début du pipeline : un scraper qui s'est cassé après une refonte de site, des requêtes qui se font bloquer, ou des pages qui s'affichent dans un navigateur mais renvoient une coquille vide à une récupération HTTP classique. Traiter l'acquisition comme fragile, et l'ensemble du pipeline hérite de cette fragilité.

Ce guide vous montre comment construire un pipeline de données web évolutif avec Crawlbase pour la collecte et des outils ETL standard pour le reste. Vous collecterez des pages avec la Crawling API pour le travail à la demande et le Crawler asynchrone pour les tâches à fort volume, vous transformerez et validerez le HTML brut, vous chargerez des lignes propres dans le stockage et planifierez l'ensemble avec supervision. Chaque étape dispose de code exécutable que vous pouvez adapter.

À quoi ressemble un pipeline de données web évolutif

Le schéma est la forme ETL classique avec une séparation des responsabilités importante. Crawlbase se situe en amont comme couche d'ingestion et gère tout ce qui rend le scraping instable : rendu JavaScript, rotation des adresses IP, routage des requêtes et atténuation des blocages. Vos systèmes gèrent le parsing, la validation, le stockage et l'analyse. Le flux se lit de gauche à droite :

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

La raison de placer la frontière ici est la durabilité. Les sites web externes ne sont pas des dépendances stables ; ils livrent des changements de mise en page, font des expériences et déploient des défenses anti-bots sans prévenir. En plaçant une couche de collecte gérée en amont, un changement de site devient une question de configuration plutôt qu'une panne du pipeline. Crawlbase vous offre deux outils de collecte pour deux formes de charge, et un pipeline de production utilise généralement les deux.

• Crawling API pour la récupération en temps réel et à la demande d'URLs connues. Vous envoyez une URL, elle renvoie la page.
• Async Crawler pour la collecte à grande échelle en mode fire-and-forget. Vous poussez des URLs, il les récupère de façon asynchrone et envoie les résultats en POST à votre webhook.

C'est la même séparation que toute opération sérieuse de scraping e-commerce finit par adopter : un chemin rapide pour les recherches ciblées et un chemin en masse pour la couverture. Si vous êtes nouveau aux mécaniques de proxy en dessous, qu'est-ce qu'un serveur proxy est un contexte utile, mais le but d'une API gérée est que vous n'ayez pas à gérer tout cela.

Étape 1 : collecter avec la Crawling API

La Crawling API prend une URL et votre token, et renvoie la page rendue. Vous envoyez un GET HTTP ; elle achemine la requête via un pool d'adresses IP rotatives, rend optionnellement JavaScript lorsque vous passez un token JS, et renvoie le HTML (ou du JSON parsé). L'appel le plus simple est un simple curl :

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

Dans un pipeline, vous voulez un collecteur petit et réutilisable plutôt qu'un curl brut. Installez le client officiel et encapsulez l'appel pour que le reste du pipeline reçoive du HTML propre et ne pense jamais aux tokens ni aux tentatives. Utilisez le token JS pour les pages rendues côté client et le token normal pour le HTML statique.

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')

Deux détails rendent ce code adapté à un pipeline plutôt qu'à un simple prototype. Premièrement, il vérifie status_code et lève une exception sur tout ce qui n'est pas une récupération propre, afin qu'une mauvaise page soit signalée bruyamment plutôt que d'empoisonner votre entrepôt avec des lignes vides. Deuxièmement, l'indicateur render garde vos sites d'appel honnêtes sur les pages qui ont besoin de JavaScript : ne payez le coût du rendu que là où le contenu le demande réellement. Ce collecteur est l'unité que votre planificateur appellera pour chaque URL connue.

Étape 2 : monter en volume avec le Crawler asynchrone

La Crawling API est synchrone : une requête, une réponse, et votre code attend. C'est exactement ce qu'il faut pour quelques centaines d'URLs connues. Pour des dizaines de milliers, bloquer sur chaque appel ne passe pas à l'échelle. Le Crawler asynchrone inverse le modèle. Vous poussez des URLs dans un crawler nommé, la requête renvoie immédiatement un ID de requête, Crawlbase récupère la page en arrière-plan, et quand c'est fait, il envoie le résultat en POST à votre endpoint de callback. Rien dans votre code ne bloque en attendant les pages.

Vous activez le mode asynchrone en ajoutant deux paramètres au même endpoint : callback=true et crawler=YourCrawlerName (vous créez le crawler une fois dans le tableau de bord et le pointez vers l'URL de votre webhook). Pousser une URL ressemble à ceci :

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

Au lieu du corps de la page, vous récupérez un ID de requête, ce qui signifie que l'URL est en file d'attente :

{ "rid": "1e92e8bf4618772871c14d4" }

De votre côté, pousser un grand lot est une boucle serrée. L'objectif est le débit : vous lancez toutes les URLs sans attendre qu'aucune ne se termine, et la file d'attente absorbe le travail.

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',
])

L'autre moitié de l'asynchrone est le gestionnaire de callback. Crawlbase envoie la page crawlée en POST au webhook que vous avez enregistré avec le crawler, envoyant le HTML dans le corps de la requête et les métadonnées (l'ID de requête, l'URL d'origine et le statut) dans les en-têtes. Votre gestionnaire doit faire le minimum : accuser réception rapidement avec un 200 et transmettre la charge utile à votre étape de transformation. Effectuer un parsing lourd en ligne risque de faire expirer la livraison et de provoquer des tentatives répétées.

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'))

Si vous préférez ne pas faire tourner de webhook du tout, pointez le crawler vers Crawlbase Cloud Storage et interrogez-le à la place ; le compromis est un petit délai contre zéro infrastructure. Dans tous les cas, le modèle asynchrone vous permet de collecter des millions de pages sans que votre application ne bloque jamais sur une récupération.

Étape 3 : transformer et valider le HTML brut

La collecte vous donne du HTML. L'étape de transformation transforme ce HTML en enregistrements propres et typés, et rejette tout ce qui ne passe pas un critère de qualité. C'est là que beaucoup de pipelines se dégradent silencieusement : un job signale un succès, mais les lignes qu'il a écrites sont vides parce qu'un sélecteur a dérivé. Validez explicitement pour qu'un échec de parsing ressemble à un échec.

Parsez avec ce qui convient à votre stack ; l'exemple utilise BeautifulSoup. La fonction extrait les champs, les normalise en types natifs et refuse d'émettre un enregistrement avec un nom manquant ou un prix non parsable.

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

La forme qui compte : nettoyez chaque champ en type natif (un prix en float, une chaîne sans espaces), abandonnez les enregistrements incomplets plutôt que d'écrire des blancs, et levez une exception quand une page entière ne produit rien, afin qu'un sélecteur dérivé soit détecté le jour même où il se casse plutôt que des semaines plus tard dans un rapport. Si vous souhaitez ignorer entièrement le parsing pour les sites pris en charge, la Crawling API renvoie directement du JSON structuré et cette étape devient un simple relais.

Étape 4 : charger dans le stockage

Avec des enregistrements validés en main, écrivez-les quelque part d'interrogeable. La destination dépend de l'échelle et de l'usage : une base de données relationnelle comme PostgreSQL pour l'accès transactionnel, un entrepôt comme BigQuery pour l'analyse, un moteur de recherche, ou une plateforme de streaming en aval. SQLite suffit pour illustrer le schéma, et le schéma est ce qui se généralise : faire un upsert sur une clé stable afin que relancer le pipeline mette à jour les lignes existantes au lieu de les dupliquer.

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()

L'upsert est ce qui rend l'étape de chargement idempotente : exécuter le même lot deux fois laisse la table dans le même état, ce qui est exactement ce que vous voulez lorsqu'un planificateur réessaie une exécution échouée. L'horodatage collected_at vous donne un signal de fraîcheur que vous utiliserez pour la surveillance à l'étape suivante. Remplacez les appels SQLite par le client de votre entrepôt et la logique reste inchangée.

Étape 5 : automatiser, planifier et surveiller

Les pièces se composent en une seule fonction de pipeline, et c'est cette fonction que votre planificateur appelle. Connecter collect, transform et load avec un try/except par URL empêche une mauvaise page de tuer toute une exécution.

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')

Pour l'exécuter selon un calendrier, l'option la plus simple est cron. Cette entrée exécute le pipeline toutes les six heures et ajoute la sortie à un journal que vous pouvez surveiller ou envoyer à votre stack de monitoring :

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

Cron convient pour quelques jobs. Dès que vous avez des dépendances entre étapes, des tentatives et des backfills, passez à un orchestrateur de workflow comme Apache Airflow ou Prefect, qui offrent des DAGs, des tentatives automatiques et une interface pour l'historique des exécutions. Pour le Crawler asynchrone, il n'y a pas du tout de planificateur à faire tourner côté collecte : vous poussez les URLs et les résultats affluent dans votre callback au fur et à mesure qu'ils se terminent.

La surveillance est ce qui distingue un pipeline de confiance d'un pipeline à surveiller en permanence. Suivez trois choses au minimum. Volume : comptages de lignes par exécution, afin qu'une chute soudaine signale un problème de collecte. Fraîcheur : les horodatages collected_at que vous avez stockés, pour pouvoir alerter quand les données deviennent obsolètes. Taux d'échec : le décompte ok/failed de chaque exécution, afin qu'une augmentation progressive vous avertisse qu'un site cible change avant que tout ne se casse. Associez cela à une bonne hygiène de scraping ; comment scraper des sites web sans être bloqué couvre les pratiques qui maintiennent la couche de collecte en bonne santé à grande échelle.

Foire aux questions

Comment construire un pipeline de données web évolutif avec Crawlbase ?

Utilisez Crawlbase comme couche de collecte et des outils ETL standard pour le reste. Collectez des pages avec la Crawling API pour les URLs connues et le Crawler asynchrone pour les gros volumes, transformez le HTML renvoyé en enregistrements typés validés, chargez-les dans le stockage avec un upsert idempotent, et planifiez l'exécution avec cron ou un orchestrateur tout en surveillant le volume, la fraîcheur et le taux d'échec. Crawlbase gère le rendu, la rotation des adresses IP et l'atténuation des blocages afin que votre code ne traite que des données propres.

Quand utiliser la Crawling API plutôt que le Crawler asynchrone ?

Utilisez la Crawling API lorsque vous avez une liste connue d'URLs et souhaitez la page immédiatement, ce qui convient aux services backend, aux jobs de monitoring et aux recherches en temps réel. Utilisez le Crawler asynchrone lorsque vous collectez à fort volume ou souhaitez une livraison fire-and-forget : vous poussez des URLs, obtenez un ID de requête instantanément, et Crawlbase envoie chaque résultat en POST à votre callback dès qu'il se termine. De nombreux pipelines utilisent les deux, l'API pour la récupération ciblée et le Crawler pour la couverture large.

Comment fonctionne le callback du Crawler asynchrone ?

Vous créez un crawler nommé dans le tableau de bord et le pointez vers l'URL de votre webhook, puis vous poussez des URLs avec callback=true et crawler=YourCrawlerName. Chaque poussée renvoie immédiatement un ID de requête. Quand Crawlbase finit de récupérer une page, il envoie un POST HTTP à votre webhook avec le HTML dans le corps et les métadonnées dans les en-têtes. Votre gestionnaire doit renvoyer un 200 rapidement et traiter la charge utile en dehors de la bande pour que la livraison n'expire pas.

Dois-je encore gérer les proxies ou les défenses anti-bots ?

Non. La Crawling API et le Crawler acheminent les requêtes via un pool d'adresses IP rotatives, rendent JavaScript lorsque vous passez un token JS et appliquent l'atténuation des blocages côté serveur. Vous envoyez une URL et récupérez une page, vous n'avez donc pas à faire tourner un pool de proxies ni une flotte de navigateurs sans tête. Si vous n'avez besoin que d'adresses IP rotatives brutes pour votre propre stack, le Smart AI Proxy expose le même réseau comme endpoint proxy standard.

Comment empêcher mon pipeline d'écrire des données vides ou mauvaises ?

Validez lors de l'étape de transformation. Vérifiez le statut de la réponse lors de la collecte et levez une exception sur tout ce qui n'est pas une récupération propre, puis lors du parsing, abandonnez les enregistrements avec des champs obligatoires manquants et levez une exception quand une page entière produit zéro enregistrement, car cela signifie généralement qu'un sélecteur a dérivé. Rendez le chargement idempotent avec un upsert pour que les tentatives ne dupliquent pas les lignes, et stockez un horodatage de collecte pour surveiller la fraîcheur et alerter quand les données deviennent obsolètes.

Ce pipeline peut-il gérer des millions de pages ?

Oui. Le goulot d'étranglement dans une conception naïve est le blocage sur chaque récupération synchrone, que le Crawler asynchrone supprime en mettant le travail en file d'attente et en livrant les résultats via callback. Poussez de grands lots sans attendre, laissez la file d'attente absorber la charge et traitez les résultats au fur et à mesure qu'ils arrivent. Pour les programmes très grands ou continus, un plan entreprise ajoute le débit et le support dont la collecte à fort volume a besoin.

IK

Ian Kalvin

Ingénieur support technique · Crawlbase

Ingénieur support technique chez Crawlbase, il écrit depuis la première ligne de ce qui casse réellement en production, sur les configurations de scraping et de proxys.

NZ

Neil Zamora

Architecte senior · Crawlbase

Architecte senior chez Crawlbase, axé sur les systèmes derrière le crawling à grande échelle : rotation de proxys, résilience anti-bot et les API qui masquent cette complexité.