Node.jsとWebスクレイピングAPIによるスケーラブルなデータパイプラインの構築

Webスクレイピングを通じて数千ページを取得するNode.jsパイプライン。並行数の制限、リトライ、再開可能な出力、テスト済みの設計。

Web Scraping APIで1ページを取得するのはコード1行で済む。しかし、プランの上限を超えず、プロセスがクラッシュしても作業を失わず、失敗を静かに見逃すこともなく、毎晩5万ページを取得するとなると、それはエンジニアリングの問題になる。Node.jsはこれに向いている。作業のほとんどはネットワーク待ちであり、それはまさにNode.jsのイベントループが得意とするところだ。

このガイドでは、ShifterのWeb Scraping APIの上に、本番運用に耐える小さなパイプラインをNode.jsで構築する。並行数の制限、適切なエラーに対するリトライ、再開可能な出力、そして独立したエラーログを備える。依存関係なしで全体は60行ほどであり、APIのモックに対してテスト済みだ。まだ最初のリクエストを行っていない場合は、Web Scraping APIの初心者向けガイドから始めるとよい。

パイプラインの構造

スクレイピングパイプラインには4つの役割がある。

キュー: 取得対象のURLの一覧。
ワーカー: キューからURLを取り出し、一定数ずつAPIを呼び出す。
エラー処理: 後で成功しうるものはリトライし、成功しえないものは記録する。
出力: クラッシュしても失われず、再実行時に完了済みの作業をスキップできる。

プロキシやブラウザ、API側での一部のリトライはAPIが面倒を見てくれる。パイプラインの仕事は、適切なレートでAPIにリクエストを送り続け、何が起きたかの信頼できる記録を保つことだ。

並行性: `Promise.all`で一括処理するのではなくワーカーを使う

最もよくあるNode.jsの間違いは、すべてのURLをリクエストにマッピングしてawait Promise.all(...)で一括処理することだ。URLが5万件あれば、これは5万件のリクエストを一斉に発射することになる。APIはそのほとんどに429 Too Many Requestsで応答する。どのプランにも並行数の上限があるからだ（Shifterではプランに応じて同時20〜500リクエスト）。

解決策は、1つのキューを共有する固定数のワーカープールだ。各ワーカーは次のURLを取り出して処理し、終わればまた次を取りに戻る。こうすることで、常にconcurrencyで指定した数ちょうどのリクエストが処理中となる。ライブラリは不要で、共有配列とループで実現できる。

リトライ: 後で成功しうるエラーに対してのみ行う

すべてのエラーがリトライに値するわけではない。タイムアウト、レンダリング失敗、429やサーバーエラーは、次の試行で成功する可能性がある。一方、不正なAPIキーや不正な形式のリクエストは永久に失敗し続けるので、リトライしても時間の無駄になる。このパイプラインは408、422、429、500に対して、試行ごとに長くなる待機時間を挟んでリトライし、それ以外はすべて即座に失敗として記録する。待機時間をなぜ長くしていくべきかについては、Webスクレイパーのリトライとバックオフのガイドで解説している。

出力: 追記専用で再開可能

成功した結果はそれぞれ、届き次第すぐに改行区切りJSON（NDJSON）ファイルに追記される。1行につき1URL分のデータとタイムスタンプが記録される。これには3つの利点がある。

クラッシュしても失われるのは処理中だったリクエストだけ。 すでに書き込まれたものはすべて残る。
再実行のコストが低い。 起動時にパイプラインは出力ファイルを読み込み、すでに記録済みのURLをスキップする。これにより、失敗した実行を再開する際に同じページ分を二重に支払わずに済む。
失敗は別途保持される。.errorsファイルに記録され、レビューしたり次回実行にフィードバックしたりできる。

再実行が同じ結果を生むように設計することは、冪等なスクレイパーの中心的な考え方であり、そのガイドでさらに詳しく扱っている。

コード

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;
}

これはNode.js組み込みのfetch、AbortSignal.timeout、fs/promisesしか使っていないため、何もインストールせずに現行のどのNode.jsリリースでも動作する。

商品ページのリストに対して実行し、HTMLではなくJSONをAPIに要求する例を示す。

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 }

テスト方法

APIのローカル代替を用意し、22件のURLを配信した。通常のページ20件、429で2回失敗した後に成功するページ1件、常に401を返すページ1件だ。テストでは以下を確認した。

21件のページが成功し、1件の失敗がエラーファイルに記録されること。
不安定なページは3回目の試行でリトライが成功し、401のページは1回しか試行されないこと。
concurrency: 5のとき、同時処理中のリクエストが5件を超えないこと。
2回目の実行で、完了済みの21件すべてがスキップされ、失敗分のみが再試行されること。

さらにスケールさせるには

上記の単一プロセス設計は、1回の実行で数万ページ規模を問題なく処理できる。それ以上の規模になる場合は次を検討する。

キューをメモリの外に出す。 データベースのテーブルやメッセージキューを使えば、複数のプロセスで作業を分担でき、マシン全体の再起動にも耐えられる。
結果を届き次第データベースに読み込む。 NDJSONは適切な中間形式であり、読み込みの手順についてはWeb Scraping APIのデータをSQLに移行するガイドで扱っている。
バックプレッシャーをかける。 取得後の処理が取得そのものより遅い場合は、結果がメモリ上に積み上がるのではなく、ワーカーの速度を落とすようにする。分散クローラーにおけるバックプレッシャーとフロー制御を参照のこと。
数値を監視する。 実行ごとの成功数、失敗数、リトライ数を記録し、失敗率が変動したら通知する。重要な指標についてはスクレイピングパイプラインの監視のガイドにまとめている。

FAQ

Node.jsはWebスクレイピングパイプラインに適していますか。

APIの上に構築するパイプラインであればそうだ。作業のほとんどはネットワーク応答待ちであり、Node.jsは1つのプロセスで少ないオーバーヘッドのまま多数の並行リクエストを扱える。CPU負荷の高いパース処理は、APIの抽出ルールか別のステップで行う方がよい。

Node.jsのスクレイパーは何件の同時リクエストを送るべきですか。

Webスクレイピング用APIプランが許す分だけであり、それ以上は送らない。その上限に設定した固定数のワーカープールを使い、429応答はペースを落とすべきという合図として扱う。

スクレイピングのジョブを再開可能にするには。

結果が届き次第、URLをキーとした追記専用のファイルやテーブルに書き込み、起動時にすでに結果があるURLはスキップする。こうすれば、クラッシュで失われるのは処理中だったリクエストの分だけになる。

キュー用のライブラリは必要ですか。

単一プロセスなら不要だ。共有配列と固定数の非同期ワーカーで十分である。複数のプロセスやマシンで作業を分担する必要が出てきたら、データベースやメッセージキューに移行する。

まとめ

Node.jsでスケーラブルなスクレイピングパイプラインを作るには4つのことが必要だ。プランの並行数に合わせた固定数のワーカープール、後で成功しうるエラーに限定したリトライ、再実行から再開できる追記専用の出力、そして失敗の独立した記録。上記のコードはこの4つすべてを約60行で実現している。APIはプロキシとブラウザを担当し、パイプラインの仕事は着実にAPIへリクエストを送り続け、何が起きたかを決して見失わないことだ。