Node.jsで分散ウェブクローラーを構築する：キューと重複排除

distributed-crawler/
├── package.json
└── src/
    └── crawler.mjs

{
  "name": "distributed-crawler-example",
  "private": true,
  "type": "module",
  "scripts": {
    "start": "node src/crawler.mjs"
  },
  "dependencies": {
    "@scrapeless-ai/sdk": "1.3.1",
    "bullmq": "5.79.3"
  },
  "engines": {
    "node": ">=22"
  }
}

import { createHash } from "node:crypto";
import { Queue, Worker } from "bullmq";
import { ScrapingCrawl } from "@scrapeless-ai/sdk";

const redis = {
  host: process.env.REDIS_HOST ?? "127.0.0.1",
  port: Number(process.env.REDIS_PORT ?? 6379)
};
const allowedHost = process.env.ALLOWED_HOST ?? "example.com";
const queueName = `crawl-${hostKey(allowedHost)}`;
const frontier = new Queue(queueName, { connection: redis });
const quarantine = new Queue(`${queueName}-quarantine`, { connection: redis });
const crawl = new ScrapingCrawl({
  apiKey: process.env.SCRAPELESS_API_KEY
});

function sha256(value) {
  return createHash("sha256").update(value).digest("hex");
}

function hostKey(host) {
  return host.toLowerCase().replaceAll(".", "-");
}

function canonicalize(input) {
  const url = new URL(input);
  url.hash = "";
  url.hostname = url.hostname.toLowerCase();
  for (const key of ["utm_source", "utm_medium", "utm_campaign"]) {
    url.searchParams.delete(key);
  }
  url.searchParams.sort();
  return url.href;
}

async function enqueue(url, crawlId, depth = 0, parentUrl = null) {
  const canonicalUrl = canonicalize(url);
  const parsed = new URL(canonicalUrl);
  if (parsed.hostname !== allowedHost) {
    throw new Error(`クローリングスコープ外のホスト: ${parsed.hostname}`);
  }

  const key = sha256(canonicalUrl);
  await frontier.add(
    "collect-page",
    {
      url: canonicalUrl,
      host: parsed.hostname,
      depth,
      crawlId,
      parentUrl,
      schemaVersion: "crawl-page-v1"
    },
    {
      jobId: key,
      removeOnComplete: 1000,
      removeOnFail: false
    }
  );
  return key;
}

const worker = new Worker(
  queueName,
  async (job) => {
    try {
      const result = await crawl.scrapeUrl(job.data.url, {
        formats: ["markdown", "links"],
        onlyMainContent: true,
        timeout: 15000
      });
      const markdown = result.markdown ?? result.data?.markdown;
      if (typeof markdown !== "string" || markdown.length < 200) {
        return { state: "rejected", reason: "コンテンツ契約", url: job.data.url };
      }

      const record = {
        key: job.id,
        url: job.data.url,
        crawlId: job.data.crawlId,
        depth: job.data.depth,

contentHash: sha256(markdown),
        markdown
      };

      console.log(JSON.stringify({ state: "accepted", ...record }));
      return { state: "accepted", key: record.key, contentHash: record.contentHash };
    } catch (error) {
      await quarantine.add("inspect-page", {
        ...job.data,
        sourceJobId: job.id,
        reason: error instanceof Error ? error.message : "unknown"
      });
      return { state: "quarantined", key: job.id };
    }
  },
  {
    connection: redis,
    concurrency: 4,
    limiter: { max: 2, duration: 1000 }
  }
);

worker.on("completed", (job, result) => {
  console.log(JSON.stringify({ event: "completed", jobId: job.id, result }));
});

worker.on("failed", (job, error) => {
  console.error(JSON.stringify({
    event: "worker-failed",
    jobId: job?.id,
    message: error.message
  }));
});

await enqueue(`https://${allowedHost}/`, "demo-crawl");

1. **クロール:** スコープ、種URL、締切、ポリシーバージョン、全体の状態。
2. **ページ:** 正規キー、ソースURL、最新の受け入れハッシュ、収集時間、スキーマバージョン。
3. **ページバージョン:** コンテンツハッシュ、ペイロードの場所、メタデータ、取得の出所。

キューは長期データベースではありません。ジョブのクリーンアップポリシーにより、完了したエントリが削除される場合がありますが、ページテーブルはアイデポテントキーとバージョン履歴を保持しなければなりません。

ページが変更されていない場合は、ペイロードを重複させずに新鮮さのタイムスタンプを更新します。変更された場合は、新しい不変のページバージョンを書き込み、ページレコードをそれに指し示し、別のイベントを通じて下流のインデクシングに通知します。

## クロールの観察

キューの長さだけでは誤解を招く可能性があります。クローラーはフロンティアを空にできながら、すべてのページを拒否することができます。以下を追跡します：

| 信号 | 答えられる質問 |
|---|---|
| 1分あたりの受け入れページ数 | 有用なデータは到着していますか？ |
| 発見から受け入れまでのラグ | パイプラインはどれほど古くなっていますか？ |
| 重複エンキュー率 | 正規化は効果的ですか？ |
| 理由別の拒否率 | ソースマークアップや検証が変更されましたか？ |
| 検疫の年齢 | 運用の負債が蓄積していますか？ |
| ホストのリクエストペース | ポリシーは遵守されていますか？ |
| コンテンツ変更率 | リフレッシュスケジュールは適切ですか？ |
| キューの年齢パーセンタイル | ワーカーのキャパシティは十分ですか？ |

OpenTelemetryは、<a href="https://opentelemetry.io/docs/concepts/signals/" rel="nofollow"><strong>テレメトリシグナルガイド</strong></a>でトレース、メトリクス、ログ、バゲージについて説明しています。プロデューサー、キューイベント、取得コール、ストレージ書き込み、および下流インデックスイベントで同じクロールIDを使用します。

過去の受け入れデータや古い検疫エントリについて警告を出しますが、プロセスのクラッシュだけではありません。プロセスは健全である一方、データセットは静かに変化が止まることがあります。

## デプロイメントチェックリスト

- ワークロードに適した永続性、認証、ネットワーク制御、およびバックアップを伴うRedisを実行します。
- ワーカーをステートレスに保ち、インスタンス間で同じイメージをデプロイします。
- APIキーをシークレットマネージャまたは環境注入に保存し、ジョペイロード内には決して保存しません。
- キューの保持ポリシーをページの保持ポリシーから独立して定義します。
- クロール深度、ページ数、時間、およびホストスコープを制限します。
- プロデューサーとワーカーが共有する一つのホストポリシーソースを使用します。
- デプロイ中にワーカーを排出して、アクティブなリースが放棄されないようにします。
- キャンセル、検疫解除、ストレージのアイデポテント性、およびソースポリシーの変更をテストします。
- Node.js、BullMQ、Scrapeless SDK、およびページスキーマのバージョンを記録します。

まず一つの承認されたホストと小さなページ制限から始めます。ダッシュボードが受け入れたデータ、新鮮さ、リクエストポリシーが目標内にあることを示すまで、ドメインを追加しません。

## 結論: 不確実性ではなくフロンティアを拡大する

分散型ウェブクローラーは、すべてのURLに1つの正規のアイデンティティがあり、すべてのホストに明確な予算があり、すべてのページが意味のある状態で終わるときに信頼できるものになります。Node.jsとBullMQはフロンティアおよびワーカーの調整を担当し、Scrapelessは管理されたレンダリングとネットワーク処理が必要な場合にページ実行を担当できます。

一つの公的で承認されたドメインを持つ限定された概念実証を作成し、[Scrapelessの価格ページ](https://www.scrapeless.com/ja/pricing?utm_source=website&utm_medium=blog&utm_campaign=crawl&utm_term=distributed-web-crawler-nodejs)を確認した後、[Scrapelessアカウントを作成](https://app.scrapeless.com/passport/login?utm_source=website&utm_medium=blog&utm_campaign=crawl&utm_term=distributed-web-crawler-nodejs)します。取得をクロールを通じてルーティングし、受け入れたページと新鮮さを測定した後にワーカー数を増やします。

## よくある質問

### ウェブクローラーを分散型にする要因は何ですか？

そのURLフロンティアとジョブ状態が複数のワーカープロセスまたはマシンで共有されていること。ワーカーは独立したジョブを請け負い、結果を共有ストレージに書き込むことができ、一つのプロセスのメモリに依存せずにスケールできます。

### Node.jsクローラーにBullMQを使用する理由は？

BullMQはRedisバックのキュー、分散ワーカー、同時実行制御、イベント、ジョブ識別子を提供します。クローラーは依然として独自のURLポリシー、コンテンツ契約、耐久性のあるストレージ、および可観測性が必要です。

### URLの重複排除はワーカー間でどのように機能しますか？

エンキューする前にURLを正規化し、正規の形式をハッシュ化し、ダイジェストをキュージョブIDおよびストレージキーとして使用します。Redisがジョブの作成を調整し、データベースはキューの保持が古いジョブを削除した後もアイデポテント性を保ちます。

### すべてのワーカーが独自のブラウザを起動するべきですか？

必ずしも必要ではありません。ローカルブラウザはコンテナのサイズ、メモリ使用量、および運用作業を増加させます。管理された取得レイヤーはレンダリングされたコンテンツを返すことができ、ワーカーはスケジューリング、検証、発見、およびストレージに集中できます。

### 失敗したページジョブはどのように処理するべきですか？

ビジネスの成果をインフラのイベントから分離します。無効なコンテンツは拒否できます。不明な実行エラーは検査と明示的なリリースのために検疫キューに入れることができます。無限の自動ループを避けます。