Webスクレイピングを個人開発で始める場合、requests と BeautifulSoup を使って同期的に処理するだけでも十分なケースがあります。
しかし、スクレイピング対象のURLが増えたり、複数ユーザーから同時にリクエストを受け付けたりすると、単純な同期処理では徐々に限界が見えてきます。
例えば、
- 大量のURLを処理したい
- 複数のスクレイピング処理を並列実行したい
- 長時間かかる処理をAPIリクエストから切り離したい
- 処理状態をデータベースで管理したい
- ワーカー間の競合を防ぎたい
- 失敗したジョブを再実行したい
といった要件が出てきます。
そこで今回は、Django REST Framework、Celery、RabbitMQ、PostgreSQL、BeautifulSoupを組み合わせた非同期スクレイピング基盤について解説します。
全体アーキテクチャ
今回想定する基本構成は以下です。
┌─────────────────┐
│ Django API │
│ DRF Layer │
└────────┬────────┘
│
│ Task enqueue
▼
┌─────────────────┐
│ RabbitMQ │
│ Message Queue │
└────────┬────────┘
│
▼
┌─────────────────┐
│ Celery Workers │
│ Distributed │
└────────┬────────┘
│
├───────────────┐
▼ ▼
┌─────────────────┐ ┌─────────────────┐
│ PostgreSQL │ │ S3 / MinIO │
│ Job State/Data │ │ Exported Data │
└─────────────────┘ └─────────────────┘
処理の流れはシンプルです。
API Request
↓
Django REST Framework
↓
Celery Task
↓
RabbitMQ
↓
Celery Worker
↓
Webページ取得
↓
BeautifulSoup
↓
データ抽出
↓
PostgreSQL / S3
APIからスクレイピング処理そのものを直接実行しないことがポイントです。
なぜ非同期処理が必要なのか
例えば、APIで以下のような処理を実装したとします。
def scrape(url):
response = requests.get(url)
soup = BeautifulSoup(response.text, "html.parser")
# データ抽出
return data
これをDjangoのAPIから直接呼び出すと、
Client
│
▼
Django API
│
├── requests.get()
├── BeautifulSoup
├── Data extraction
│
▼
Response
となります。
Webサイトによってはレスポンスに数秒以上かかることがあります。
さらに1000件、10000件とURLが増えると、APIサーバーがスクレイピング処理に占有されてしまいます。
そこで、スクレイピング処理をCeleryに移します。
Client
│
▼
Django API
│
└── Celery Task
│
▼
RabbitMQ
│
▼
Celery Worker
│
▼
Scraping
APIはジョブを登録したらすぐにレスポンスを返すことができます。
Django REST Frameworkでジョブを登録する
例えば、以下のようなAPIを考えます。
POST /api/scraping/jobs/
リクエスト：
{
"url": "https://example.com/products"
}
Django REST FrameworkではSerializerを利用できます。
from rest_framework import serializers
class ScrapingJobSerializer(serializers.Serializer):
url = serializers.URLField()
ViewではCelery Taskを呼び出します。
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework import status
from .serializers import ScrapingJobSerializer
from .tasks import scrape_url
class ScrapingJobView(APIView):
def post(self, request):
serializer = ScrapingJobSerializer(
data=request.data
)
serializer.is_valid(
raise_exception=True
)
url = serializer.validated_data["url"]
task = scrape_url.delay(url)
return Response(
{
"task_id": task.id,
"status": "PENDING"
},
status=status.HTTP_202_ACCEPTED
)
ここで重要なのは、
scrape_url.delay(url)
です。
delay()によってCeleryに処理を渡し、API自身はスクレイピング完了を待ちません。
Celery Workerでスクレイピングする
Celery側では、
from celery import shared_task
import requests
from bs4 import BeautifulSoup
@shared_task
def scrape_url(url: str):
response = requests.get(
url,
headers={
"User-Agent": "Mozilla/5.0"
},
timeout=10
)
response.raise_for_status()
soup = BeautifulSoup(
response.text,
"html.parser"
)
title = soup.select_one("h1")
return {
"url": url,
"title": (
title.get_text(strip=True)
if title
else None
)
}
という処理を実装できます。
これで、
Django
↓
RabbitMQ
↓
Celery
↓
Requests
↓
BeautifulSoup
という非同期スクレイピングパイプラインが完成します。
PostgreSQLでジョブの状態を管理する
実際のシステムでは、CeleryのTask IDだけに状態管理を依存するのではなく、アプリケーション側でもジョブの状態を管理すると扱いやすくなります。
例えば、
class ScrapingJob(models.Model):
STATUS_CHOICES = [
("PENDING", "Pending"),
("RUNNING", "Running"),
("SUCCESS", "Success"),
("FAILED", "Failed"),
]
url = models.URLField(
max_length=2048
)
status = models.CharField(
max_length=20,
choices=STATUS_CHOICES,
default="PENDING"
)
result = models.JSONField(
null=True,
blank=True
)
error_message = models.TextField(
null=True,
blank=True
)
created_at = models.DateTimeField(
auto_now_add=True
)
updated_at = models.DateTimeField(
auto_now=True
)
これによって、
PENDING
↓
RUNNING
↓
SUCCESS
または、
PENDING
↓
RUNNING
↓
FAILED
という状態を管理できます。
ステートマシンとして考える
スクレイピング処理が複雑になると、単純な status だけでは管理しづらくなります。
例えば、
URL_FINDING
↓
URL_CRAWLING
↓
DATA_EXTRACTION
↓
VALIDATION
↓
COMPLETED
のように、複数のステップを持つパイプラインを考えることができます。
その場合は、現在のステップと各ステップの状態をDBに保存します。
class InputState(models.Model):
base_url = models.URLField(
max_length=2048
)
user_instruction = models.TextField()
schema_instruction = models.TextField()
site_type = models.CharField(
max_length=20,
choices=[
("general", "General"),
("ec", "E-Commerce"),
]
)
auto_flow = models.BooleanField(
default=True
)
current_step = models.CharField(
max_length=50
)
url_finder_status = models.CharField(
max_length=20,
default="PENDING"
)
url_crawler_status = models.CharField(
max_length=20,
default="PENDING"
)
extraction_status = models.CharField(
max_length=20,
default="PENDING"
)
これにより、システム全体の状態をDBから確認できます。
Celeryワーカーの競合状態
非同期処理を導入すると、別の問題が発生します。
それが**競合状態（Race Condition）**です。
例えば、
InputState
status = RUNNING
version = 5
というデータがあったとします。
Worker AとWorker Bが同時にこのデータを取得した場合、
Worker A ──┐
├── version = 5
Worker B ──┘
となります。
Worker Aが更新して、
version = 6
になったあと、Worker Bが古い version = 5 を使って更新すると、Worker Aの更新を上書きしてしまう可能性があります。
これはLost Updateと呼ばれる典型的な競合問題です。
楽観的ロックを導入する
この問題を解決する方法の一つが、Optimistic Locking（楽観的ロック）です。
モデルにversionを持たせます。
class VersionedModel(models.Model):
version = models.IntegerField(
default=0
)
class Meta:
abstract = True
更新時には、
UPDATE ...
WHERE id = ?
AND version = ?
という条件で更新します。
例えば、
affected = (
InputState.objects
.filter(
pk=state.pk,
version=state.version
)
.update(
status="RUNNING",
version=models.F("version") + 1
)
)
更新された行数を確認します。
if not affected:
raise ConcurrencyError(
"Version conflict"
)
これによって、別のWorkerが先に更新していた場合には競合を検出できます。
なぜ悲観的ロックではなく楽観的ロックなのか
もちろんPostgreSQLには、
select_for_update()
を利用した悲観的ロックもあります。
しかし、大量のスクレイピング処理では長時間DBロックを保持することは避けたいケースがあります。
特に、
DB transaction
↓
HTTP request
↓
External website
↓
数秒待機
↓
DB update
のような処理でDBロックを維持すると、他のトランザクションに影響を与える可能性があります。
そのため、
外部サイトへのアクセスとDBのロックを分離する
という設計が重要になります。
大量URLを効率的に登録する
例えば1万件のURLを登録するとします。
単純に、
for url in urls:
ScrapingURL.objects.create(
url=url
)
とすると、1件ごとにDBへのINSERTが発生します。
大量データでは非効率です。
そこでDjango ORMの bulk_create() を利用します。
objects = [
ScrapingURL(
input_state_id=state_id,
url=url,
status="PENDING"
)
for url in urls
]
ScrapingURL.objects.bulk_create(
objects,
batch_size=1000
)
これによって大量のINSERTをまとめて処理できます。
重複URLへの対応
スクレイピングシステムでは、同じURLが複数回登録される可能性があります。
例えば、
https://example.com/a
https://example.com/b
https://example.com/a
のようなケースです。
この場合、DB側でも一意性を担保しておくと安全です。
class ScrapingURL(models.Model):
input_state = models.ForeignKey(
InputState,
on_delete=models.CASCADE
)
url = models.URLField(
max_length=2048
)
status = models.CharField(
max_length=20,
default="PENDING"
)
class Meta:
constraints = [
models.UniqueConstraint(
fields=[
"input_state",
"url"
],
name="unique_state_url"
)
]
そして、
ScrapingURL.objects.bulk_create(
objects,
batch_size=1000,
ignore_conflicts=True
)
とすることで、重複によるINSERTエラーを回避できます。
スクレイピング処理とDB処理を分離する
ここまでの内容を組み合わせると、より実用的なパイプラインになります。
Django API
│
▼
Create Job
│
▼
PostgreSQL
│
▼
Celery
│
▼
RabbitMQ
│
▼
┌────────────────┐
│ Celery Worker │
└───────┬────────┘
│
▼
HTTP Fetch
│
▼
BeautifulSoup
│
▼
Extraction
│
▼
Validation
│
┌─────┴─────┐
▼ ▼
PostgreSQL S3
このように責任を分離することで、各コンポーネントを独立してスケールできます。
Celery Workerを増やす
例えばWorkerが1台しかない場合、
RabbitMQ
│
▼
Worker 1
となります。
大量のURLを処理する場合は、
RabbitMQ
┌──────┼──────┐
▼ ▼ ▼
Worker 1 Worker 2 Worker 3
のようにWorkerを増やすことができます。
Celeryを利用することで、スクレイピング処理を水平スケールしやすくなります。
ただし、Workerを増やせば無条件に高速になるわけではありません。
対象サイトへのアクセス頻度が高すぎると、
- Rate Limit
- HTTP 429
- Server overload
- IP blocking
などが発生する可能性があります。
そのため、実運用ではConcurrencyだけでなく、対象サイトごとのRate Limitも考慮する必要があります。
S3 / MinIOで大量データを保存する
スクレイピングしたデータが大量になる場合、すべてをPostgreSQLのJSONFieldに保存する必要はありません。
例えば、
PostgreSQL
→ Job metadata
→ Status
→ URL
→ Processing result
を保存し、
S3 / MinIO
→ Raw HTML
→ JSON export
→ CSV export
→ Large result files
を保存するという分離ができます。
この構成にすると、DBをジョブ管理に集中させることができます。
実運用を意識したディレクトリ構成
Djangoプロジェクトでは、例えば以下のような構成にできます。
project/
│
├── config/
│ ├── settings.py
│ ├── celery.py
│ └── urls.py
│
├── scraping/
│ ├── models.py
│ ├── serializers.py
│ ├── views.py
│ ├── tasks.py
│ │
│ ├── fetchers/
│ │ ├── requests.py
│ │ └── playwright.py
│ │
│ ├── parsers/
│ │ └── beautifulsoup.py
│ │
│ ├── extractors/
│ │ ├── product.py
│ │ └── article.py
│ │
│ ├── validators/
│ │ └── product.py
│ │
│ └── exceptions.py
│
└── manage.py
例えば、
fetchers
はWebページの取得、
parsers
はHTML解析、
extractors
は必要なデータの抽出、
validators
はデータの検証、
tasks
はCeleryによる非同期処理、
という責任分担にできます。
まとめ
Webスクレイピングは、最初は
Requests
+
BeautifulSoup
だけでも十分です。
しかし、処理するURLが増えたり、本番環境で継続的に実行したりする場合には、非同期処理と状態管理が重要になります。
今回紹介した構成では、
Django REST Framework
↓
RabbitMQ
↓
Celery
↓
Requests / Playwright
↓
BeautifulSoup
↓
Validation
↓
PostgreSQL / S3
というパイプラインを構築できます。
また、大規模なスクレイピングでは、
- 非同期処理
- Workerの水平スケール
- PostgreSQLによる状態管理
- Optimistic Locking
- Bulk Insert
- URLの重複排除
- Rate Limit
- Retry
- データ検証
- Raw DataとMetadataの分離
などを考慮する必要があります。
特に重要なのは、「スクレイピング処理」と「ジョブ管理」を分離することです。
BeautifulSoup自体は非常にシンプルなライブラリですが、その周辺に適切なアーキテクチャを構築することで、大量のURLを扱う非同期スクレイピング基盤へ発展させることができます。
まずは、
Requests
→ BeautifulSoup
→ Pydantic
というシンプルな構成から始め、必要になったタイミングで、
Django
→ Celery
→ RabbitMQ
→ PostgreSQL
→ S3
へ段階的に拡張していくのがおすすめです。