はじめに

システム開発第一事業部の大泉です。普段はエンジニアとして Python でバックエンドの開発を行っています。

AWS の SQS を使用して 1500 程度のサイトを対象に 1 日数百万ページを処理するクローラーを構築しました。開発初期、ページ数の多いサイトの URL でキューが埋め尽くされ、他のサイトのクローリングがほとんど行われない問題が発生しました。本記事ではこのようなキュー詰まりを解消し、すべてのサイトを平準化してクローリングする方法をご紹介します。

なお、サンプルコードは実装のイメージが伝わりやすいように簡略化したものであり、実際の実装とは異なります。

アーキテクチャ

SQS でクローリング対象の URL をキューイングするアーキテクチャです。1 つのメッセージに 1 つの URL を含めるようにします。

ECS: クローリング実行
SQS: URL をキューイング
Lambda: トリガーとなる URL をキューに送信
EventBridge Scheduler: Lambda を定期的に呼び出す
ElastiCache for Valkey: 単位時間あたりのクローリング回数を保持

ECS で行うクローリングの処理フローです。

クローリング処理フロー

キューにポーリングを行う
受信したメッセージから URL を取り出し、フェッチを行う
取得した HTML からリンクをすべて抽出する
抽出したリンクを 1 メッセージ 1 URL として SQS に送信する
1 に戻る

クローラーの立ち上げ時はキューにメッセージが何も入っていないため、トリガーとなるメッセージが必要です。それを投入するのが Lambda です。例えば以下のようなフローになります。

Lambda はサイト情報テーブルから https://example.com/ を取得
https://example.com/ をキューに送信
クローリング処理フロー の 1 に進む

キュー詰まり

問題になるのがリンクが多いサイトです。1 URL から大量の URL が抽出され、さらにそこから大量の URL が抽出され、というように一部のサイトの URL でキューが埋め尽くされます。その結果、他のサイトの URL を取り出すのに非常に時間がかかります。そこで以下を導入しました。

キューの分離
同時実行制限
可視性タイムアウト
遅延キュー

キューの分離

「何回目のフェッチで見つかった URL か」をクローリングの深さとします。例えばクローリングのトリガーとなる Lambda によってキューに送信された URL は深さ 0 です。深さ 0 の URL から取得した URL は深さ 1 です。

クローリングの深さによって優先度高キューと優先度低キューの 2 つに分離しました。優先度高キューには深さが小さい URL を入れます。優先度低キューには深さが大きい URL を入れます。

例えば、深さ 0 と 1 は優先度高キュー、深さ 2 以降は優先度低キューに入れるようにします。

優先度高キューのポーリングを先に行い、メッセージを受信しなければ優先度低キューのポーリングを行います。これにより深さが小さい URL のクローリングが優先して行われるようになります。

同時実行制限

同時実行制限は次の 2 種類を導入しました。

サーバ単位での同時実行制限: 1 サーバ内でのサイトごとのクローリング回数を平準化
時間単位での同時実行制限: 1 時間あたりのサイトごとのクローリング回数を平準化

2 についてはキャッシュサーバ（ElastiCache for Valkey）にメッセージ処理回数を保存します。キャッシュサーバを共有ストレージとして利用することで、各サーバ間でサイトごとのメッセージ処理回数を共有できるようになります。それぞれの同時実行制限に引っかかったメッセージは、次に述べる「メッセージ単位での可視性タイムアウト」を利用し、一定期間受信できないようにします。

サーバ単位での同時実行制限

サーバ内でサイトごとのメッセージ処理回数を管理するため、専用のクラスを実装しました。辞書型（マップ）で {"siteA": 1, "siteB": 2} のようにサイトごとのメッセージ処理回数を保持します。

以下はサンプルコードです。

class ConcurrencyManager:
    """サイトごとの同時実行数を管理"""

    def __init__(self):
        self.active_tasks = defaultdict(int)

    def can_acquire(self, site_id: int, limit: int) -> bool:
        """上限を超えていないかチェック"""
        if self.active_tasks[site_id] < limit:
            self.active_tasks[site_id] += 1
            return True
        return False

    def release(self, site_id: int):
        """クローリングが完了（メッセージ処理が完了）したときはデクリメント"""
        if self.active_tasks[site_id] > 0:
            self.active_tasks[site_id] -= 1

時間単位での同時実行制限

キャッシュサーバに処理回数を保存し、1 時間経過後にキャッシュが消えるように有効期限を設定しています（ttl_seconds = 3600）。

以下はサンプルコードです。制限を超えている場合は呼び出し元に False を返し、クローリングを行わないようにします。

def check_limit(site_id: int, limit: int) -> bool:
    """サイトごとのメッセージ処理回数をチェック"""
    key = f"count:{site_id}"
    ttl_seconds = 3600

    with client.pipeline(transaction=True) as pipe:
        pipe.incr(key)
        pipe.ttl(key)
        count, ttl = pipe.execute()

        # TTLが設定されていない場合、有効期限を設定
        if ttl == -1:
            client.expire(key, ttl_seconds)

        # 制限を超えているかチェック
        return count <= limit

メッセージ単位での可視性タイムアウト

可視性タイムアウトとは、あるコンシューマーが処理中のメッセージを他のコンシューマーが重複処理しないようにするため、一定期間キューから見えなくする機能です。SQS ではキュー自体の可視性タイムアウトとメッセージ単位での可視性タイムアウトを設定できます。

docs.aws.amazon.com

ここではメッセージ単位での可視性タイムアウトを利用します。同時実行制限に引っかかったサイトの URL に対してフェッチを行わず、可視性タイムアウトを延長してキューに戻します。これにより、制限に引っかかったメッセージが見えなくなっている時間が長くなり、他サイトのメッセージを先に取り出せるようになります。

docs.aws.amazon.com

可視性タイムアウトの設定時間は、あらかじめ決めた範囲（例: 5～10 秒）の中でランダムに決定します。ランダムに設定することで同じ条件のメッセージが同時に見えるようになることを防ぎます。

一斉にメッセージを処理するため、boto3 ライブラリの change_message_visibility_batch を使用して可視性タイムアウトを変更します。

boto3.amazonaws.com

以下はサンプルコードです。

entries = []
for i, receipt_handle in enumerate(receipt_handles):
    # 最小設定時間と最大設定時間を決定
    min_delay, max_delay = self.delay_calculator.determine_visibility_timeout()
    # ランダムに可視性タイムアウトを決定
    entries.append({
        "Id": str(i),
        "ReceiptHandle": receipt_handle,
        "VisibilityTimeout": random.randint(min_delay, max_delay)
    })
# バッチ(1度に10メッセージ)で可視性タイムアウト変更
for i in range(0, len(entries), 10):
    batch_entries = entries[i : i + 10]
    async with self._create_client() as client:
        await client.change_message_visibility_batch(
            QueueUrl=queue_url,
            Entries=batch_entries
        )

メッセージ単位での遅延キュー

遅延キューは、メッセージが初めてキューに入ってから一定期間キューから見えなくなる機能です。つまり、すぐにメッセージが取り出せないようにします。これを使うことで、クローリング回数の多いサイトの処理を抑制し、回数の少ないサイトのメッセージを優先的に取り出せるようにします。

docs.aws.amazon.com

なお、可視性タイムアウトはコンシューマーがメッセージを受信してから見えなくなるのに対し、遅延キューはメッセージがキューに入ってから見えなくなります。

フェッチレスポンスから抽出した URL （次のクローリング URL）をキューに送信するときに、以下の条件によって遅延キューの設定時間を決定します。

クローリングの深さ
サイトごとのメッセージ処理数によるパーセンタイル（メッセージ処理数が多いサイトを特定する）

これらを考慮し、最小設定時間～最大設定時間の間でランダムに設定します（ここでも揺らぎを持たせることで、同時にメッセージが取り出されるのを防ぎます）。これにより、リンクが多いサイトでは階層が深くなるほど新たに見つかる URL の遅延が長くなり、埋もれていた小規模サイトの URL を取り出しやすくなります。

例）

クローリングの深さ 1、サイトのメッセージ処理回数によるパーセンタイル 50% → 1 分～2 分
クローリングの深さ 2、サイトのメッセージ処理回数によるパーセンタイル 80% → 10 分～15 分

可視性タイムアウトと同様に、一斉にメッセージを処理するため、boto3 ライブラリの send_message_batch を使用して遅延キューを設定します。

boto3.amazonaws.com

以下はサンプルコードです。

for i in range(0, len(urls), 10):
    # クローリングで見つかったURLをバッチ送信のため分割
    batch_urls = urls[i : i + 10]

    # クローリングの深さとサイトごとのメッセージ処理数から最小設定時間と最大設定時間を決定
    min_delay_seconds, max_delay_seconds = self.delay_calculator.determine_delay_seconds()

    # キューに送信するためのエントリーを作成. 遅延キューの設定はDelaySeconds
    entries = []
    for idx, url in enumerate(batch_urls):
        base_site_info.url = url
        entries.append({
            "Id": str(idx),
            "MessageBody": json.dumps(base_site_info.to_dict(), ensure_ascii=False),
            "DelaySeconds": random.randint(min_delay_seconds, max_delay_seconds),
        })

    # バッチ(1度に10メッセージ)でキューに送信
    async with self._create_client() as client:
        await client.send_message_batch(
            QueueUrl=self.crawler_low_queue_url,
            Entries=entries,
        )

まとめ

SQS を利用した大規模クローラーにおけるキュー詰まりの問題とその解決策について紹介しました。クローリング平準化のため以下の 4 つのアプローチを組み合わせました。

キューの分離: 優先して処理したいメッセージとそうでないメッセージを分離する（本記事では、クローリングの深さによって分離した）
同時実行制限: 各サーバ内部で保持する同時実行数と、サーバ全体で共有する処理回数を管理することにより、クローリング回数の多いサイトの実行数を制限し、その分少ないサイトのクローリングを実行する
可視性タイムアウト: 同時実行制限に引っかかったメッセージの可視性タイムアウトを延長し、クローリング回数の少ないサイトのメッセージを取り出しやすくする
遅延キュー: クローリングの深さとサイトのメッセージ処理回数に応じて新規 URL の遅延を調整し、クローリング回数の少ないサイトのメッセージを取り出しやすくする

以上のように、SQS で大規模なメッセージを扱う際は単に流し込むだけでなく流量制御が重要になります。本記事が同様の課題を抱える方の参考になれば幸いです。

参考

docs.aws.amazon.com

boto3.amazonaws.com

テコテックの採用活動について

テコテックでは新卒採用、中途採用共に積極的に募集をしています。
採用サイトにて会社の雰囲気や福利厚生、募集内容をご確認いただけます。
ご興味を持っていただけましたら是非ご覧ください。

tecotec.co.jp

はじめに
OpenSearch とは
- 全文検索とは
主要な概念
環境構築
- Docker で OpenSearch を起動する
- Python クライアントのインストール
- クライアントの接続
インデックスの作成（マッピング定義）
- text と keyword の違い
CRUD
- ドキュメントの登録（Create）
- ドキュメントの取得（Read）
- ドキュメントの更新（Update）
- ドキュメントの削除（Delete）
検索（Search）
- 全件取得
- 全文検索（match）
- 複数条件の組み合わせ（bool クエリ）
- ソート
まとめ
参考
テコテックの採用活動について

はじめに

システム開発第一事業部の大泉です。普段はエンジニアとして Python でバックエンドの開発を行っています。この記事では、OpenSearch を知らない方向けに OpenSearch の概要と Python を使った CRUD 操作・検索について解説します。

OpenSearch とは

OpenSearch は分散型の検索・分析エンジンです。Elasticsearch のオープンソースのフォークとして開発されています。

aws.amazon.com

github.com

主な用途は次のとおりです。

全文検索: EC サイトの商品検索、社内ドキュメント検索
ログ分析: アクセスログ・アプリケーションログのリアルタイム集計
時系列データの可視化: アクセス数の推移やエラー率などをグラフ・チャートで可視化するダッシュボードを構築

テコテック開発者ブログ

AIで作業は「速く」なるのに、仕事は「早く」ならない理由

はじめに

大規模クローリングにおけるSQSキュー詰まりの解消法

はじめに

アーキテクチャ

キュー詰まり

キューの分離

同時実行制限

サーバ単位での同時実行制限

時間単位での同時実行制限

メッセージ単位での可視性タイムアウト

メッセージ単位での遅延キュー

まとめ

参考

テコテックの採用活動について

AIのトークンは「使うほど良い」のか　～探索と定型化2つのフェーズで考える～

はじめに

社内アプリコンテスト本編「いざ、本戦！🔥11チームの熱きピッチと、最高の懇親会」

効率化の前に、あえてAI抜きでやってみるということ

はじめに

AI時代のエンジニアに問われる「説明する力」

はじめに

【OpenSearch】ゼロから始める OpenSearch — Python で学ぶインデックス操作と全文検索

はじめに

OpenSearch とは