ScrapelessとDuckDBを使用してスクレイピングデータ分析パイプラインを構築する

pip install "duckdb==1.5.4"

export SCRAPELESS_API_KEY="your_api_key_here"

import json, os, urllib.request
from html.parser import HTMLParser

API = "https://api.scrapeless.com/api/v2/unlocker/request"

def fetch(url: str) -> str:
    payload = json.dumps({
        "actor": "unlocker.webunlocker",
        "input": {"url": url, "js_render": True, "headless": True},
    }).encode()
    req = urllib.request.Request(
        API, data=payload,
        headers={"x-api-token": os.environ["SCRAPELESS_API_KEY"],
                 "Content-Type": "application/json"},
    )
    with urllib.request.urlopen(req, timeout=120) as r:
        return json.loads(r.read())["data"]

class QuoteParser(HTMLParser):
    def __init__(self):
        super().__init__()
        self.rows, self._cur, self._cap = [], None, None
    def handle_starttag(self, tag, attrs):
        a = dict(attrs); cls = a.get("class", "")
        if tag == "div" and "quote" in cls:
            self._cur = {"text": "", "author": "", "tags": []}
        elif self._cur is not None and tag == "span" and "text" in cls:
            self._cap = "text"
        elif self._cur is not None and tag == "small" and "author" in cls:
            self._cap = "author"
        elif self._cur is not None and tag == "a" and "tag" in cls:
            self._cap = "tag"
    def handle_data(self, data):
        if self._cap == "text": self._cur["text"] += data
        elif self._cap == "author": self._cur["author"] += data
        elif self._cap == "tag": self._cur["tags"].append(data.strip())
    def handle_endtag(self, tag):
        if self._cap in ("text", "author", "tag"): self._cap = None
        if tag == "div" and self._cur and self._cur["text"]:
            self.rows.append(self._cur); self._cur = None

rows = []
for page in range(1, 4):
    p = QuoteParser(); p.feed(fetch(f"https://quotes.toscrape.com/page/{page}/"))
    rows.extend({"page": page, **r} for r in p.rows)

print(f"取得したページ数: 3 | 抽出したレコード数: {len(rows)}")
with open("quotes.jsonl", "w", encoding="utf-8") as f:
    for r in rows: f.write(json.dumps(r, ensure_ascii=False) + "\n")
print(f"quotes.jsonlに書き込みました ({os.path.getsize('quotes.jsonl')} バイト)")

取得したページ数: 3 | 抽出したレコード数: 30
quotes.jsonlに書き込みました (7690 バイト)

import duckdb

con = duckdb.connect("quotes.duckdb")

con.execute("""
    CREATE OR REPLACE TABLE quotes AS
    SELECT
        page::INTEGER            AS page,
        text                     AS quote_text,
        author,
        tags,
        len(tags)                AS tag_count
    FROM read_json_auto('quotes.jsonl')
""")

total = con.sql("SELECT count(*) FROM quotes").fetchone()[0]
authors = con.sql("SELECT count(DISTINCT author) FROM quotes").fetchone()[0]
print(f"読み込まれた行数: {total} | 異なる著者数: {authors}")

print(con.sql("""
    SELECT author, count(*) AS quotes, round(avg(tag_count), 2) AS avg_tags
    FROM quotes
    GROUP BY author
    ORDER BY quotes DESC, author
    LIMIT 5
""").to_df().to_string(index=False))

con.execute("COPY quotes TO 'quotes.parquet' (FORMAT PARQUET, COMPRESSION ZSTD)")
import os
print(f"parquetバイト数: {os.path.getsize('quotes.parquet')} | jsonlバイト数: {os.path.getsize('quotes.jsonl')}")

rt = duckdb.sql("SELECT count(*) AS n, count(DISTINCT author) AS a FROM 'quotes.parquet'").fetchone()
print(f"パーケットからの往復 -> 行数: {rt[0]}, 著者数: {rt[1]}")

読み込まれた行数: 30 | 異なる著者数: 20
         著者      引用   平均タグ数
アルバート・アインシュタイン       6      2.83
   J.K.ローリング       3      1.33
     ボブ・マーリー       2      1.00
      ドクター・スース       2      2.00
 マリリン・モンロー       2      4.00
parquetバイト数: 4674 | jsonlバイト数: 7690
パーケットからの往復 -> 行数: 30, 著者数: 20

import json, os, urllib.request
from html.parser import HTMLParser

API = "https://api.scrapeless.com/api/v2/unlocker/request"

def fetch(url: str) -> str:
    payload = json.dumps({
        "actor": "unlocker.webunlocker",
        "input": {"url": url, "js_render": True, "headless": True},
    }).encode()
    req = urllib.request.Request(
        API, data=payload,

取得したページ数: 3 | 抽出されたレコード数: 30
読み込まれた行数: 30 | 異なる著者数: 20
Parquet バイト数: 4674 | JSONL バイト数: 7690
Parquet からの往復 -> 行数: 30, 著者数: 20

DuckDBはサーバーなし、アカウントなし、ネットワークの往復なしでプロセス内で動作し、実際にほとんどのスクレイピングプロジェクトが操作するスケールに適しています。JSONとParquetをネイティブに読み込むため、ローダーを記述する必要はありません。クラウドウェアハウスは複数のチームが同じテーブルに同時アクセスする必要があるときにその地位を得ます — パイプラインが自身の出力に対してSQLを必要とする場合ではありません。

**Q: ロードの前にタグリストのようなネストされたフィールドをフラット化する必要がありますか？**

いいえ、フラット化は情報を失います。DuckDBはリストタイプをエンドツーエンドでサポートしているので、`tags`はロード、SQLの`len(tags)`、Parquetの書き込みと読み戻しを通じて配列のままです。区切られた文字列に圧縮すると、その後のすべてのクエリが再解析を強いられます。

**Q: スクレイピングとロードの間にJSON Linesを書く理由は何ですか？**

それは二つの半分を分離します。スクレイピングは遅く、失敗しやすい部分です；レコードがディスクに一行ずつ保存されたら、再取得せずに何度でもロードと変換を再実行できます。一行ずつ追加することは、処理が途中で終了した場合でも、以前のレコードがそのまま保たれ、読みやすい状態になることを意味します。

**Q: ParquetはJSON Linesよりどのくらい小さいですか？**

この30レコードのサンプルでは、4,674バイト対7,690バイト — おおよそ40%小さいです。このサイズでその比率を過度に解釈しないでください、ファイルのオーバーヘッドが支配的なためです。Parquetの本当の利点はカラムリードです：五つのカラムのうち二つに触れるクエリはその二つだけを読み込むため、ファイルがメモリ内に快適に収まらなくなるボリュームでは重要です。

**Q: Parquetファイルをデータベースにロードせずにクエリできますか？**

はい、ロード段階の最後の行は正にそれを行います — データベース接続を開かず、インポートなしで`SELECT ... FROM 'quotes.parquet'`を実行します。これがParquetをスクレイピングパイプラインの良い最終アーティファクトにする理由です：出力はDuckDBや他の分析エンジンによって、その場にある限りクエリ可能なままです。