Skalierbare Datenpipelines mit Node.js und Web-Scraping-APIs erstellen

Eine Seite mit einer Web Scraping API abzurufen ist eine einzige Codezeile. Fünfzigtausend Seiten jede Nacht abzurufen, ohne die Limits des eigenen Plans zu überschreiten, ohne Arbeit zu verlieren, wenn der Prozess abstürzt, und ohne Fehler stillschweigend zu verwerfen, ist ein technisches Problem. Node.js eignet sich dafür gut: Die Arbeit besteht fast ausschließlich aus Warten auf das Netzwerk, und genau das beherrscht Nodes Event Loop gut.

Diese Anleitung baut eine kleine, produktionsreife Pipeline in Node.js auf Basis von Shifters Web Scraping API: begrenzte Nebenläufigkeit, Wiederholungsversuche für die richtigen Fehler, fortsetzbare Ausgabe und ein separates Fehlerprotokoll. Das Ganze umfasst etwa sechzig Zeilen ohne Abhängigkeiten, und wir haben es gegen eine Mock-Version der API getestet. Wenn Sie noch keine erste Anfrage gestellt haben, beginnen Sie mit unserer Einführung für Einsteiger in eine Web Scraping API.

Der Aufbau der Pipeline

Eine Scraping-Pipeline hat vier Aufgaben:

1. Eine Warteschlange mit abzurufenden URLs.
2. Worker, die URLs aus der Warteschlange nehmen und die API aufrufen, eine feste Anzahl gleichzeitig.
3. Fehlerbehandlung, die wiederholt, was später erfolgreich sein kann, und aufzeichnet, was nicht kann.
4. Eine Ausgabe, die Abstürze übersteht und einem erneuten Durchlauf erlaubt, bereits erledigte Arbeit zu überspringen.

Die API kümmert sich auf ihrer Seite um Proxys, Browser und einige Wiederholungsversuche. Die Aufgabe der Pipeline ist es, sie mit der richtigen Rate zu versorgen und ein zuverlässiges Protokoll darüber zu führen, was passiert ist.

Nebenläufigkeit: Worker statt Promise.all über alles

Der häufigste Node.js-Fehler besteht darin, jede URL auf eine Anfrage abzubilden und await Promise.all(...) über die gesamte Menge auszuführen. Bei fünfzigtausend URLs löst das fünfzigtausend Anfragen gleichzeitig aus. Die API beantwortet die meisten davon mit 429 Too Many Requests, weil jeder Plan ein Limit für gleichzeitige Anfragen hat (bei Shifter je nach Plan zwischen 20 und 500 gleichzeitigen Anfragen).

Die Lösung ist ein fester Pool von Workern, die sich eine Warteschlange teilen. Jeder Worker nimmt die nächste URL, verarbeitet sie und holt sich dann die nächste, sodass jederzeit genau concurrency Anfragen gleichzeitig unterwegs sind. Dafür wird keine Bibliothek benötigt; ein gemeinsames Array und eine Schleife reichen aus.

Wiederholungsversuche: nur für Fehler, die später erfolgreich sein können

Nicht jeder Fehler verdient einen Wiederholungsversuch. Ein Timeout, ein fehlgeschlagenes Rendering, ein 429 oder ein Serverfehler können beim nächsten Versuch erfolgreich sein. Ein falscher API-Schlüssel oder eine fehlerhafte Anfrage werden dagegen für immer fehlschlagen, und ein Wiederholungsversuch verschwendet nur Zeit. Die Pipeline wiederholt 408, 422, 429 und 500 mit einer Pause, die bei jedem Versuch wächst, und verzeichnet alles andere sofort als Fehlschlag. Unsere Anleitung zu Retry und Backoff für Web Scraper erklärt, warum die Pause wachsen sollte.

Ausgabe: nur anfügend und fortsetzbar

Jedes erfolgreiche Ergebnis wird sofort nach Eintreffen an eine zeilengetrennte JSON-Datei (NDJSON) angehängt: eine Zeile pro URL mit den Daten und einem Zeitstempel. Das hat drei Vorteile:

• Ein Absturz verliert höchstens die gerade unterwegs befindlichen Anfragen. Alles bereits Geschriebene bleibt erhalten.
• Erneute Durchläufe sind günstig. Beim Start liest die Pipeline die Ausgabedatei und überspringt jede URL, die dort schon vorhanden ist, sodass Sie einen fehlgeschlagenen Durchlauf neu starten können, ohne dieselben Seiten doppelt zu bezahlen.
• Fehlschläge werden getrennt aufbewahrt, in einer .errors-Datei, die Sie überprüfen oder in den nächsten Durchlauf zurückführen können.

Die Gestaltung so, dass ein erneuter Durchlauf das gleiche Ergebnis liefert, ist der Kerngedanke hinter idempotenten Scrapern, die das Thema ausführlicher behandeln.

Der Code

import { appendFile, readFile } from 'node:fs/promises';

const API = process.env.SCRAPER_API_URL ?? 'https://scrape.shifter.io/v1';
const RETRY = new Set([408, 422, 429, 500]);
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));

// Fetch one URL through the Web Scraping API, retrying only errors that can succeed later.
export async function fetchPage(url, { extractRules, attempts = 3, backoffMs = 2000 } = {}) {
  const params = new URLSearchParams({ api_key: process.env.SHIFTER_SCRAPER_KEY, url });
  if (extractRules) params.set('extract_rules', JSON.stringify(extractRules));
  for (let attempt = 1; ; attempt++) {
    const res = await fetch(`${API}?${params}`, { signal: AbortSignal.timeout(120_000) });
    if (res.ok) return extractRules ? res.json() : res.text();
    if (!RETRY.has(res.status) || attempt === attempts) {
      throw new Error(`HTTP ${res.status} for ${url}`);
    }
    await sleep(backoffMs * attempt);
  }
}

// URLs already written to the output file, so a restarted run skips them.
async function alreadyDone(outFile) {
  try {
    const lines = (await readFile(outFile, 'utf8')).split('\n').filter(Boolean);
    return new Set(lines.map((line) => JSON.parse(line).url));
  } catch {
    return new Set();
  }
}

// Run `concurrency` workers over the queue; each result becomes one NDJSON line.
export async function runPipeline(urls, { outFile, concurrency = 10, extractRules, backoffMs } = {}) {
  const done = await alreadyDone(outFile);
  const queue = urls.filter((u) => !done.has(u));
  const stats = { skipped: urls.length - queue.length, ok: 0, failed: 0 };

  async function worker() {
    for (let url = queue.shift(); url !== undefined; url = queue.shift()) {
      try {
        const data = await fetchPage(url, { extractRules, backoffMs });
        await appendFile(outFile, JSON.stringify({ url, data, at: new Date().toISOString() }) + '\n');
        stats.ok++;
      } catch (err) {
        await appendFile(`${outFile}.errors`, JSON.stringify({ url, error: err.message }) + '\n');
        stats.failed++;
      }
    }
  }

  await Promise.all(Array.from({ length: concurrency }, worker));
  return stats;
}

Es verwendet nur die eingebauten Node-Funktionen fetch, AbortSignal.timeout und fs/promises, sodass es auf jeder aktuellen Node.js-Version läuft, ohne dass etwas installiert werden muss.

Ausführung anhand einer Liste von Produktseiten, wobei die API nach JSON statt HTML gefragt wird:

import { readFile } from 'node:fs/promises';
import { runPipeline } from './pipeline.mjs';

const urls = (await readFile('urls.txt', 'utf8')).split('\n').filter(Boolean);
const stats = await runPipeline(urls, {
  outFile: 'products.ndjson',
  concurrency: 20,                       // stay under your plan's concurrency limit
  extractRules: {
    name: { selector: 'h1', output: 'text' },
    price: { selector: '.price', output: 'text' },
  },
});
console.log(stats);                      // { skipped, ok, failed }

Wie wir es getestet haben

Wir haben die Pipeline gegen einen lokalen Ersatz für die API laufen lassen, der 22 URLs bediente: zwanzig normale Seiten, eine, die zweimal mit 429 fehlschlug, bevor sie erfolgreich war, und eine, die immer 401 zurückgab. Der Test prüfte, dass:

• 21 Seiten erfolgreich waren und ein Fehlschlag in der Fehlerdatei verzeichnet wurde;
• die instabile Seite wiederholt wurde und beim dritten Versuch erfolgreich war, während die 401-Seite nur einmal versucht wurde;
• bei concurrency: 5 nie mehr als fünf Anfragen gleichzeitig unterwegs waren;
• ein zweiter Durchlauf alle 21 fertigen Seiten übersprang und nur den Fehlschlag erneut versuchte.

Weitere Skalierung

Das oben beschriebene Einzelprozess-Design bewältigt bequem Zehntausende Seiten pro Durchlauf. Darüber hinaus:

• Verlagern Sie die Warteschlange aus dem Arbeitsspeicher. Eine Datenbanktabelle oder eine Message Queue erlaubt mehreren Prozessen, sich die Arbeit zu teilen, und übersteht Neustarts der gesamten Maschine.
• Laden Sie Ergebnisse bei Eintreffen in eine Datenbank. NDJSON ist ein gutes Landeformat; unsere Anleitung zum Übertragen von Web Scraping API-Daten nach SQL behandelt den Ladevorgang.
• Wenden Sie Backpressure an. Wenn der Schritt nach dem Abruf langsamer ist als der Abruf selbst, lassen Sie ihn die Worker bremsen, statt Ergebnisse im Arbeitsspeicher anhäufen zu lassen. Siehe Backpressure und Flusskontrolle in verteilten Crawlern.
• Behalten Sie die Zahlen im Blick. Verfolgen Sie Erfolge, Fehlschläge und Wiederholungsversuche pro Durchlauf und alarmieren Sie, wenn sich die Fehlerrate verändert. Unsere Anleitung zum Überwachen einer Scraping-Pipeline listet die wichtigen Metriken auf.

FAQ

Ist Node.js gut für Web-Scraping-Pipelines geeignet?

Ja, für Pipelines, die auf einer API aufbauen. Die Arbeit besteht größtenteils aus dem Warten auf Netzwerkantworten, und Node.js verarbeitet viele gleichzeitige Anfragen in einem Prozess mit geringem Overhead. CPU-intensives Parsen erledigt man besser über die Extraktionsregeln der API oder in einem separaten Schritt.

Wie viele gleichzeitige Anfragen sollte ein Node.js-Scraper stellen?

So viele, wie Ihr Scraping-API-Plan erlaubt, und nicht mehr. Verwenden Sie einen festen Pool von Workern, der auf dieses Limit eingestellt ist, und behandeln Sie 429-Antworten als Signal, langsamer zu werden.

Wie mache ich einen Scraping-Job fortsetzbar?

Schreiben Sie jedes Ergebnis sofort nach Eintreffen in eine nur anfügende Datei oder Tabelle, die nach URL indiziert ist, und überspringen Sie beim Start jede URL, für die bereits ein Ergebnis vorliegt. Ein Absturz kostet dann nur die Anfragen, die gerade unterwegs waren.

Brauche ich eine Queue-Bibliothek?

Für einen einzelnen Prozess nicht. Ein gemeinsames Array mit einer festen Anzahl asynchroner Worker reicht aus. Wechseln Sie zu einer Datenbank oder Message Queue, wenn mehrere Prozesse oder Maschinen sich die Arbeit teilen müssen.

Fazit

Eine skalierbare Scraping-Pipeline in Node.js braucht vier Dinge: einen festen Pool von Workern, dimensioniert nach der Nebenläufigkeit Ihres Plans, Wiederholungsversuche nur für Fehler, die später erfolgreich sein können, nur anfügende Ausgabe, von der ein erneuter Durchlauf fortsetzen kann, und ein separates Protokoll der Fehlschläge. Der obige Code erledigt all das in etwa sechzig Zeilen. Die API kümmert sich um Proxys und Browser; die Aufgabe Ihrer Pipeline ist es, sie gleichmäßig zu versorgen und nie den Überblick darüber zu verlieren, was passiert ist.