Xây dựng một quy trình phân tích dữ liệu thu thập bằng Scrapeless và DuckDB

TL;DR:

• Một pipeline thu thập dữ liệu kết thúc bằng tệp JSON chỉ là một nửa của pipeline; các câu hỏi phân tích xuất hiện sau khi dữ liệu được lưu lại.
• DuckDB đọc trực tiếp JSON Lines, vì vậy các bản ghi thu thập trở thành một bảng có thể truy vấn mà không cần máy chủ, không cần di chuyển lược đồ và không cần công cụ ETL.
• Pipeline này thu thập 3 trang thông qua API Thu thập Dữ liệu Toàn cầu Scrapeless, trích xuất 30 bản ghi, tải chúng vào DuckDB và ghi dưới dạng Parquet.
• Parquet với nén ZSTD lưu trữ cùng 30 bản ghi đó trong 4.674 byte so với 7.690 byte của JSON Lines, và DuckDB truy vấn tệp trực tiếp mà không cần tải lại.
• Mỗi giai đoạn bên dưới chạy trên máy của bạn mà không cần tài khoản kho đám mây và không cần thông tin xác thực nào ngoài khóa Scrapeless.
• Bắt đầu với kế hoạch miễn phí của Scrapeless và chỉ định giai đoạn thu thập đến nguồn của riêng bạn.

Tổng quan về Pipeline

Luồng dữ liệu bao gồm năm giai đoạn, và quyết định thiết kế thú vị là nơi mà dữ liệu ngừng là văn bản và bắt đầu trở thành một bảng.

fetch (Scrapeless) → khám phá bản ghi → trích xuất trường → chuyển đổi thành bảng kiểu (DuckDB) → lưu trữ dưới dạng Parquet

JSON Lines là định dạng chuyển giao giữa nửa thu thập dữ liệu và nửa phân tích dữ liệu. Nó thân thiện với việc bổ sung, sống sót sau một lần chạy bị lỗi mà không làm hỏng những gì đã xảy ra trước đó, và DuckDB đọc nó một cách tự nhiên — vì vậy không cần phải viết trình tải. Định dạng được quy định tại đặc tả JSON Lines.

Không có gì ở đây cần tài khoản kho. DuckDB chạy trong quy trình, điều này làm cho đây là cách rẻ nhất để có SQL thực sự trên dữ liệu đã thu thập. Khi địa điểm là một kho được quản lý, hướng dẫn nạp dữ liệu Snowflake sẽ đề cập đến cùng một giai đoạn thu thập với một khu vực hạ cánh khác.

Điều kiện tiên quyết

• Python 3.9 trở lên.
• Một khóa API Scrapeless từ bảng điều khiển.
• duckdb đã được cài đặt:

pip install "duckdb==1.5.4"

Đặt khóa:

export SCRAPELESS_API_KEY="your_api_key_here"

Giai đoạn 1–3: Thu thập, Khám phá, Trích xuất

Một cuộc gọi cho mỗi trang qua API Thu thập Dữ liệu Toàn cầu Scrapeless trả về HTML đã được render, và một bộ phân tích thư viện chuẩn biến mỗi khối trích dẫn thành một bản ghi. Trình trích xuất phát ra một đối tượng JSON mỗi dòng:

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"pages fetched: 3 | records extracted: {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"wrote quotes.jsonl ({os.path.getsize('quotes.jsonl')} bytes)")

pages fetched: 3 | records extracted: 30
wrote quotes.jsonl (7690 bytes)

Hai lựa chọn ở đây đáng để giữ trong phiên bản của riêng bạn.
Số trang được gán cho mỗi bản ghi tại thời điểm trích xuất. Nguồn gốc gần như miễn phí để ghi lại tại thời điểm thu thập và đắt đỏ để tái tạo sau đó, và chính nó cho phép bạn trả lời "trang nào đã lấy cái này" mà không cần chạy lại bất kỳ điều gì.

ensure_ascii=False giữ nguyên các ký tự trích dẫn đánh máy mà không thoát chúng. Điều này quan trọng khi văn bản là dữ liệu — đầu ra được thoát vẫn hợp lệ JSON, nhưng nó làm phình to tập tin và khiến việc kiểm tra sau này trở nên khó khăn hơn.

Giai Đoạn 4: Chuyển Đổi Thành Bảng Có Kiểu

DuckDB đọc tệp JSON Lines trực tiếp. read_json_auto suy luận sơ đồ, và câu lệnh SELECT bên cạnh là nơi bạn áp dụng các kiểu và cột suy diễn mà bạn thực sự muốn:

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"rows loaded: {total} | distinct authors: {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 bytes: {os.path.getsize('quotes.parquet')} | jsonl bytes: {os.path.getsize('quotes.jsonl')}")

rt = duckdb.sql("SELECT count(*) AS n, count(DISTINCT author) AS a FROM 'quotes.parquet'").fetchone()
print(f"round-trip from parquet -> rows: {rt[0]}, authors: {rt[1]}")

rows loaded: 30 | distinct authors: 20
         author  quotes  avg_tags
Albert Einstein       6      2.83
   J.K. Rowling       3      1.33
     Bob Marley       2      1.00
      Dr. Seuss       2      2.00
 Marilyn Monroe       2      4.00
parquet bytes: 4674 | jsonl bytes: 7690
round-trip from parquet -> rows: 30, authors: 20

Cột tags vẫn giữ dạng danh sách thay vì bị phẳng thành chuỗi phân tách. DuckDB mang theo các kiểu lồng nhau vào Parquet, vì vậy len(tags) hoạt động trong SQL và mảng tồn tại sau vòng tròn — việc làm phẳng thành "a,b,c" ở giai đoạn này là một thói quen giảm thiểu cần phải bỏ.

CREATE OR REPLACE TABLE làm cho việc tải lên trở nên idempotent. Chạy lại giai đoạn này sẽ xây dựng lại bảng từ tệp hiện tại thay vì thêm các bản sao, đây là hành vi bạn muốn khi bạn vẫn đang cải thiện bộ trích xuất.

Tổng hợp là điểm chính của toàn bộ bài tập: 30 bản ghi, 20 tác giả khác nhau, và một tác giả chiếm 6 trong số đó. Câu hỏi đó là một truy vấn một dòng chống lại một bảng và một vòng lặp gây khó chịu chống lại một tệp JSON.

Sẵn sàng chạy điều này chống lại một nguồn mà bạn quan tâm? Tạo một tài khoản Scrapeless miễn phí và thay thế URL trong giai đoạn thu thập.

Giai Đoạn 5: Lưu Dưới Dạng Parquet

Cùng 30 bản ghi đó chiếm 4,674 byte dưới dạng Parquet nén ZSTD chống lại 7,690 byte dưới dạng JSON Lines. Trên mẫu này, việc tiết kiệm là khiêm tốn; lý do để quan tâm là định dạng hoạt động ở quy mô lớn và những gì nó cho phép sau đó.

Parquet là định dạng cột và lưu trữ giá trị của từng cột cùng với mã hóa riêng của nó, được định nghĩa trong đặc tả định dạng tệp Apache Parquet. Một truy vấn chạm vào hai trong số năm cột chỉ đọc hai cột đó. Bộ mã hóa nén được sử dụng ở đây được chỉ định trong tiêu chuẩn nén Zstandard.

Dòng vòng tròn là bài kiểm tra của giai đoạn. Đọc lại tệp Parquet trả về 30 hàng và 20 tác giả khác nhau, khớp với bảng mà nó đã đến — điều này đáng được khẳng định trong bất kỳ pipeline nào ghi lại một tệp mà hệ thống khác sẽ đọc.

Lưu ý rằng truy vấn cuối cùng đọc 'quotes.parquet' như một bảng trực tiếp, không có bước nhập khẩu và không cần kết nối mở tới tệp cơ sở dữ liệu. Đó là lý do thực tế để kết thúc một pipeline thu thập dưới dạng Parquet: đầu ra có thể được truy vấn bởi DuckDB, và bởi hầu hết các công cụ phân tích khác, chính nơi nó nằm.

Pipeline Hoàn Chỉnh

Chạy như một tập lệnh, năm giai đoạn đủ ngắn để đọc trong một lần. Đây là phiên bản để sao chép:

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,
```text
Các trang đã được lấy: 3 | Số bản ghi được trích xuất: 30
Số hàng đã tải: 30 | Tác giả khác nhau: 20
Kích thước parquet: 4674 | Kích thước jsonl: 7690
Vòng lặp từ parquet -> số hàng: 30, tác giả: 20

Định hướng tiếp theo cho Pipeline này

Chia theo ngày thu thập mỗi khi bạn chạy lại nó. Việc ghi vào định dạng data/dt=<ngày-thu-thập>/quotes.parquet cho phép một truy vấn bỏ qua toàn bộ thư mục thay vì quét lịch sử.

Giữ lại các tệp JSON Lines. Chúng là bản ghi thô của những gì đã được thu thập; Parquet là sản phẩm được khai thác và có kiểu dữ liệu. Khi xảy ra lỗi trong bộ extractor, bạn sẽ tái tạo từ các tệp thô thay vì phải thu thập lại.

Thêm các câu lệnh xác nhận giữa các giai đoạn. Kiểm tra số lượng giữa trích xuất và tải sẽ phát hiện ra một bộ chọn đã dừng khớp một cách im lặng — chế độ lỗi sẽ xuất hiện như một sự giảm nhẹ số hàng mà không thông báo trong nhiều tuần sau đó.

Trước khi hướng dẫn điều này tới một nguồn trực tiếp, hãy kiểm tra điều khoản của nó và các chỉ thị trong /robots.txt, tuân theo chuẩn Giao thức loại trừ Robot. Giữ việc thu thập cho các trang công khai và trong một phạm vi trang được giới hạn như ví dụ trên.

Kết luận

Khoảng cách giữa một trình thu thập (scraper) và một pipeline phân tích nhỏ hơn nhiều so với vẻ bề ngoài. JSON Lines như một giao diện, DuckDB như một động cơ truy vấn, và Parquet như sản phẩm lưu trữ đã bao phủ nó với một phụ thuộc và không cần cơ sở hạ tầng — lấy 3 trang, xuất 30 hàng có kiểu dữ liệu, có thể truy vấn tại chỗ.

Bước mà hầu hết các pipeline bỏ qua là bước cuối cùng. Ghi vào Parquet và sau đó đọc lại để xác nhận các số lượng khớp biến bước lưu trữ thành một bước đã được xác minh, điều này tạo ra sự khác biệt giữa một tệp mà bạn có và một tệp mà bạn có thể tin tưởng.

Bắt đầu với kế hoạch miễn phí của Scrapeless để thực hiện giai đoạn lấy dữ liệu từ các mục tiêu của riêng bạn, và xem xét giá cả Scrapeless khi bạn xác định một công việc định kỳ.

Câu hỏi thường gặp

Q: Tại sao lại chọn DuckDB thay vì một kho dữ liệu đám mây cho dữ liệu thu thập?

DuckDB chạy trong quá trình mà không có máy chủ, không cần tài khoản và không có vòng lặp mạng, điều này phù hợp với quy mô mà hầu hết các dự án scraping thực sự hoạt động. Nó đọc JSON và Parquet một cách tự nhiên, vì vậy không cần phải viết bộ tải. Một kho dữ liệu đám mây có vị trí của nó khi nhiều nhóm cần truy cập đồng thời vào cùng một bảng — không phải khi một pipeline cần SQL trên đầu ra của chính nó.

**Q: Tôi có cần làm phẳng các trường lồng nhau như danh sách thẻ trước khi tải không?**

Không, và việc làm phẳng sẽ mất thông tin. DuckDB hỗ trợ các loại danh sách từ đầu đến cuối, vì vậy `tags` vẫn là một mảng trong quá trình tải, qua `len(tags)` trong SQL, và qua việc ghi và đọc lại Parquet. Việc nén chúng thành một chuỗi phân cách buộc mọi truy vấn sau phải phân tích lại.

**Q: Tại sao lại viết JSON Lines giữa việc scraping và tải lên?**

Nó tách rời hai nửa. Việc scraping là phần chậm và dễ thất bại; khi các bản ghi đã ở trên đĩa với một bản mỗi dòng, bạn có thể chạy lại quá trình tải và chuyển đổi nhiều lần mà không cần phải lấy lại. Thêm dòng theo dòng cũng có nghĩa là một lần chạy chết giữa chừng để lại các bản ghi trước đó nguyên vẹn và có thể đọc được.

**Q: Parquet nhỏ hơn bao nhiêu so với JSON Lines?**

Trên mẫu 30 bản ghi này, 4,674 byte so với 7,690 — nhỏ hơn khoảng 40%. Đừng quá chú ý vào tỷ lệ đó ở kích thước này, nơi mà chi phí tập tin chiếm ưu thế. Lợi thế thực sự của Parquet là việc đọc theo cột: một truy vấn chạm vào hai trong năm cột chỉ đọc những cột đó, điều này quan trọng ở những khối lượng mà tệp không còn vừa vặn trong bộ nhớ.

**Q: Tôi có thể truy vấn tệp Parquet mà không cần tải nó vào cơ sở dữ liệu trước không?**

Có, và dòng cuối cùng của giai đoạn tải làm chính xác điều đó — `SELECT ... FROM 'quotes.parquet'` mà không cần kết nối cơ sở dữ liệu mở và không cần nhập. Đó là lý do làm cho Parquet trở thành một sản phẩm cuối tốt cho một pipeline scraping: đầu ra vẫn có thể truy vấn nơi nó nằm, bởi DuckDB và các engine phân tích khác.