はじめに
Webスクレイピングは、データ収集の実務において広く実践されている手法です。
一方で、安定的に運用しようとすると以下のように問題が発生しがちなテーマでもあります。
- 「サイトの構造が変わって、パース処理を全部書き直し」
- 「Pythonの仮想環境が壊れて、動かなくなった」
- 「定期実行のcronがいつの間にか止まっていた」
- 「スクレイピング専用のインスタンスを管理するのが面倒」
こうした運用負荷を軽減するアプローチとして、 BigQuery Remote Functions × Playwright × TROCCO の組み合わせを紹介します。
この構成を組んでみると、以下のようなシンプルな指示で運用が可能になります。
-- スクレイピング対象のURLを登録
INSERT INTO targets (url) VALUES ('https://books.toscrape.com/');
-- SQLを実行すると、構造化されたJSONがテーブルに保存される
SELECT url, data FROM results;
-- → [{"title": "A Light in the Attic", "price": "£51.77", "in_stock": true}, ...]
URLを登録してSQLを実行すると、ブラウザが起動し、HTMLが取得され、AIが構造化し、BigQueryに保存されます。
本記事では、この「SQLだけで運用が完結するAIスクレイピング基盤」の構築方法を、実際のコードとともに解説していきます。
対象読者
- BigQueryを利用しているデータエンジニア、アナリティクスエンジニア
- スクレイピングの保守に課題を感じている人
- Python環境の管理をしたくない、SQL中心で業務を回したい人
この記事で学べること
- BigQuery Remote Functionsを使ったCloud Run連携の実装
- Playwrightによるスクレイピング
- Gemini（ML.GENERATE_TEXT）を使ったHTML構造化
- TROCCOによるワークフロー自動化
全体アーキテクチャ
┌─────────────────────────────────────────────────────────────────────────┐
│ データパイプライン │
├─────────────────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────┐ ┌──────────────┐ ┌────────────┐ ┌─────────────┐ │
│ │ TROCCO │───▶│ BigQuery │───▶│ Cloud Run │───▶│ Target Site │ │
│ │ (定期実行) │ │ (SQL実行) │ │(Playwright)│ │ (HTML) │ │
│ └──────────┘ └──────────────┘ └────────────┘ └─────────────┘ │
│ │ │ │ │ │
│ │ │ │ │ │
│ │ ▼ ▼ │ │
│ │ ┌──────────────┐ ┌────────────┐ │ │
│ │ │ Gemini │◀────│ HTML │◀─────────┘ │
│ │ │ (構造化AI) │ │ (生データ) │ │
│ │ └──────────────┘ └────────────┘ │
│ │ │ │
│ │ ▼ │
│ │ ┌──────────────┐ │
│ └────────▶│ 結果テーブル │ │
│ │ (構造化JSON) │ │
│ └──────────────┘ │
│ │
└─────────────────────────────────────────────────────────────────────────┘
各コンポーネントの役割
| コンポーネント | 役割 |
|---|---|
| TROCCO | ワークフローのスケジューリング、エラー監視、リトライ制御 |
| BigQuery | SQLによるオーケストレーション、Remote Functionの呼び出し |
| Cloud Run | Playwrightによるヘッドレスブラウザ実行 |
| Gemini | HTMLからの情報抽出、構造化JSON生成 |
Cloud RunもGeminiも、BigQueryのSQL関数として呼び出す形にしています。
TROCCOがそのSQLを定期実行し、エラー時には通知してくれます。
実装ステップ1：Cloud Run (Playwright) の準備
まずは、ブラウザを動かすCloud Runサービスを構築します。
Pythonコード
FastAPIとPlaywrightを組み合わせたWebスクレイピングサービスです。BigQuery Remote Functionsが要求するリクエスト/レスポンス形式に準拠しています。
main.py
"""
BigQuery Remote Functions用 Webスクレイピングサービス
"""
import re
from typing import Any
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from playwright.async_api import async_playwright, Browser
browser: Browser | None = None
@asynccontextmanager
async def lifespan(app: FastAPI):
"""アプリケーション起動時にブラウザを初期化"""
global browser
playwright = await async_playwright().start()
browser = await playwright.chromium.launch(headless=True)
yield
if browser:
await browser.close()
await playwright.stop()
app = FastAPI(lifespan=lifespan)
class BigQueryRequest(BaseModel):
requestId: str
caller: str
sessionUser: str
calls: list[list[Any]]
class BigQueryResponse(BaseModel):
replies: list[str | None]
def clean_html(html: str) -> str:
"""script/styleタグを除去してHTMLを軽量化"""
html = re.sub(r'<script[^>]*>[\s\S]*?</script>', '', html, flags=re.IGNORECASE)
html = re.sub(r'<style[^>]*>[\s\S]*?</style>', '', html, flags=re.IGNORECASE)
html = re.sub(r'\s+', ' ', html)
return html.strip()
async def fetch_page(url: str) -> str:
"""URLからHTMLを取得"""
page = await browser.new_page()
try:
await page.goto(url, wait_until="networkidle", timeout=30000)
return clean_html(await page.content())
finally:
await page.close()
@app.post("/")
async def handle(request: BigQueryRequest) -> BigQueryResponse:
"""BigQueryからのリクエストを処理"""
replies = []
for call in request.calls:
try:
html = await fetch_page(call[0])
replies.append(html[:500_000]) # トークン節約のため500KB制限
except Exception:
replies.append(None)
return BigQueryResponse(replies=replies)
URLを受け取り、Playwrightでページを取得し、HTMLを返すだけの単機能サービスになっています。
requirements.txt
fastapi==0.109.0
uvicorn==0.27.0
playwright==1.41.2
pydantic==2.5.3
Dockerfile
# Playwrightの公式イメージを使用（ブラウザ同梱）
FROM mcr.microsoft.com/playwright/python:v1.41.2-jammy
WORKDIR /app
ENV PORT=8080
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY main.py .
# 非rootユーザーで実行（セキュリティ対策）
USER pwuser
EXPOSE 8080
CMD ["python", "main.py"]
Cloud Runへのデプロイ
# プロジェクトIDを設定
PROJECT_ID="your-project-id"
REGION="asia-northeast1"
SERVICE_NAME="bq-scraper"
# Container Registryにビルド＆プッシュ
gcloud builds submit --tag gcr.io/${PROJECT_ID}/${SERVICE_NAME}
# Cloud Runにデプロイ
gcloud run deploy ${SERVICE_NAME} \
--image gcr.io/${PROJECT_ID}/${SERVICE_NAME} \
--region ${REGION} \
--platform managed \
--memory 2Gi \
--cpu 2 \
--timeout 300 \
--concurrency 10 \
--min-instances 0 \
--max-instances 5 \
--no-allow-unauthenticated
実装ステップ2：BigQuery Remote Functionの定義
Cloud Runサービスができたら、BigQueryからそれを呼び出すためのRemote Functionを定義します。
Step 2-1：CLOUD RESOURCE接続の作成
まず、BigQueryがCloud Runを呼び出すための接続を作成します。
# 接続の作成
bq mk --connection \
--connection_type=CLOUD_RESOURCE \
--location=asia-northeast1 \
scraper_connection
Google Cloudコンソールから作成する場合は、「BigQuery」→「外部接続」→「接続を追加」から「Cloud Resource」を選択します。
Step 2-2：サービスアカウントへの権限付与
接続作成時に生成されるサービスアカウントに、Cloud Runを呼び出す権限を付与します。
# 接続のサービスアカウントを取得
CONNECTION_SA=$(bq show --format=json --connection \
${PROJECT_ID}.asia-northeast1.scraper_connection \
| jq -r '.cloudResource.serviceAccountId')
# Cloud Run起動権限を付与
gcloud run services add-iam-policy-binding bq-scraper \
--region=asia-northeast1 \
--member="serviceAccount:${CONNECTION_SA}" \
--role="roles/run.invoker"
Step 2-3：Remote Function DDL
SQLからスクレイピングを呼び出す関数を定義します。
-- データセット作成（未作成の場合）
CREATE SCHEMA IF NOT EXISTS `your-project.scraping`
OPTIONS (location = 'asia-northeast1');
-- Remote Function定義
CREATE OR REPLACE FUNCTION `your-project.scraping.fetch_html`(url STRING)
RETURNS STRING
REMOTE WITH CONNECTION `your-project.asia-northeast1.scraper_connection`
OPTIONS (
endpoint = 'https://bq-scraper-XXXXX-an.a.run.app',
max_batching_rows = 10
);
Step 2-4：動作確認
books.toscrape.com はスクレイピング練習用に公開されているサイトです（サイトに "We love being scraped!" と明記されています）
-- 書籍サイトからHTMLを取得
SELECT scraping.fetch_html('https://books.toscrape.com/') AS html;
これで、SQLを実行するだけでWebページのHTMLが取得できるようになりました。
実装ステップ3：Geminiでの構造化とTROCCOでの運用
HTMLを取得できるようになったら、次はAIで構造化し、定期実行の仕組みを整えます。
Step 3-1：Geminiモデルの登録
BigQueryからGeminiを呼び出すためのモデルを登録します。
# Vertex AI接続の作成
bq mk --connection \
--connection_type=CLOUD_RESOURCE \
--location=asia-northeast1 \
gemini_connection
接続に紐づくサービスアカウントに、Vertex AI の利用権限を付与します。
# 接続のサービスアカウントを取得
GEMINI_SA=$(bq show --format=json --connection \
${PROJECT_ID}.asia-northeast1.gemini_connection \
| jq -r '.cloudResource.serviceAccountId')
# Vertex AI 権限を付与
gcloud projects add-iam-policy-binding ${PROJECT_ID} \
--member="serviceAccount:${GEMINI_SA}" \
--role="roles/aiplatform.user"
-- Geminiモデルの登録
CREATE OR REPLACE MODEL `your-project.scraping.gemini_model`
REMOTE WITH CONNECTION `your-project.asia-northeast1.gemini_connection`
OPTIONS (
endpoint = 'gemini-2.5-flash'
);
Step 3-2：URL + 自然言語指示 から 構造化データへの変換
URLと「何を抽出したいか」を指定することで、構造化されたJSONが得られます。
まず、スクレイピング対象を管理するテーブルを作成します。
-- ターゲットURLを管理するテーブル
CREATE TABLE IF NOT EXISTS `your-project.scraping.targets` (
url STRING
);
-- books.toscrape.com（スクレイピング練習用サイト）を使った例
INSERT INTO `your-project.scraping.targets` VALUES
('https://books.toscrape.com/'),
('https://books.toscrape.com/catalogue/page-2.html');
次に、スクレイピングとAI構造化を実行するクエリです。
-- スクレイピング → AI構造化 → 保存を1クエリで実行
CREATE OR REPLACE TABLE `your-project.scraping.results` AS
-- Step 1: スクレイピング結果を確定
WITH scraped AS (
SELECT
url,
scraping.fetch_html(url) AS html
FROM `your-project.scraping.targets`
),
-- Step 2: Gemini で構造化
raw_results AS (
SELECT
url,
CURRENT_TIMESTAMP() AS scraped_at,
JSON_VALUE(ml_generate_text_result, '$.candidates[0].content.parts[0].text') AS raw_text
FROM ML.GENERATE_TEXT(
MODEL `your-project.scraping.gemini_model`,
(
SELECT
url,
CONCAT(
'以下のHTMLから情報を抽出し、JSON配列で出力してください。\n',
'抽出項目: 書籍のタイトル（title）、価格（price）、在庫状況（in_stock）\n',
'出力形式: [{"title": "...", "price": "...", "in_stock": true/false}]\n',
'JSON以外は出力しないでください。\n\n',
html
) AS prompt
FROM scraped
WHERE html IS NOT NULL
),
STRUCT(0.1 AS temperature, 65535 AS max_output_tokens)
)
)
-- Step 3: Markdownコードフェンスを除去してJSONパース
SELECT
url,
scraped_at,
SAFE.PARSE_JSON(
TRIM(REPLACE(REPLACE(raw_text, '```json', ''), '```', ''))
) AS data
FROM raw_results;
実行すると、このような結果が得られます。
SELECT url, data FROM `your-project.scraping.results`;
| url | data |
|---|---|
| https://books.toscrape.com/ | [{"title": "A Light in the Attic", "price": "£51.77", "in_stock": true}, {"title": "Tipping the Velvet", "price": "£53.74", "in_stock": true}, ...] |
サイトのDOM構造などを意識せずとも、自然言語の指示によって解釈ができるため、Pythonに慣れていなくてもある程度の精度調整ができそうです。
Step 3-3：TROCCOでのワークフロー設定
TROCCOを使って、このSQLを定期実行し、エラー監視します。
[データ転送ジョブ: スクレイピング実行]
│
▼
[データチェック: 取得件数検証]
│
├── 成功 → [完了通知]
│
└── 失敗 → [Slack通知] → [手動確認待ち]
TROCCOでの設定手順
1. データマートジョブの作成
- サービス: BigQuery（自由記述モード）
- 実行SQL: 上記のCREATE TABLE文から1行目を削除した、SELECT文
2. ワークフローの設定
- スケジュール: 毎日9:00 など（または必要な頻度）
3. データチェックの追加
-- 取得件数が期待値を下回っていないかチェック
SELECT
COUNT(*) AS success_count,
(SELECT COUNT(*) FROM `your-project.scraping.targets`) AS total_count
FROM
`your-project.scraping.results`
WHERE
scraped_at > TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 1 DAY)
AND data IS NOT NULL
4. 通知設定
- 成功時: Slackに完了通知
- 失敗時: Slack + メールでアラート
TROCCOを使うことで、GUI上でのスケジュール管理や履歴管理、通知管理などが可能です。
実践的なTips
Tip 1：HTMLサイズの最適化
Geminiへの入力トークンを節約するため、必要な部分だけを抽出します。
-- 特定のセクションだけを抽出するパターン
SELECT
url_id,
REGEXP_EXTRACT(html_content, r'<main[^>]*>(.*?)</main>') AS main_content
FROM
scraped_results;
Tip 2：レート制限への対応
大量のURLを処理する場合は、バッチ処理で分割します。
-- 100件ずつ処理
DECLARE batch_size INT64 DEFAULT 100;
DECLARE offset INT64 DEFAULT 0;
WHILE offset < (SELECT COUNT(*) FROM target_urls) DO
-- バッチ処理
CREATE OR REPLACE TABLE scraped_results_batch AS
SELECT ...
FROM target_urls
LIMIT batch_size OFFSET offset;
SET offset = offset + batch_size;
END WHILE;
Tip 3：エラーハンドリング
スクレイピング失敗時のフォールバック処理です。
SELECT
url_id,
CASE
WHEN html_content IS NULL THEN 'SCRAPE_FAILED'
WHEN structured_data IS NULL THEN 'PARSE_FAILED'
ELSE 'SUCCESS'
END AS status,
structured_data
FROM
scraped_results;
まとめ
本記事では、BigQuery Remote Functions × Playwright × TROCCOを組み合わせた「SQLだけで運用が完結するAIスクレイピング基盤」の構築方法を紹介しました。
一度Pythonのサンプルコードをデプロイしておけば、SQLによる実行制御で
- ① Remote Functionsによってブラウザを操り
- ② Geminiによってテキストを理解し
- ③ TROCCOによって運用が自動化されます
既存のデータ基盤にスクレイピング処理を追加したい方などは、ぜひ検討してみてください。