大规模将 Web Scraping API 数据导入 SQL 数据库

一个网络抓取 API 消除了采集中最难的部分：代理、渲染、重试和封锁。它没有消除的是响应到达之后发生的一切，而这恰恰是大多数管道真正失败的地方。

这些失败是悄无声息的。没人去重的重试产生了重复的行。一个价格列全是像 "$16.99" 这样的字符串，被人在仪表盘里强行转成了数字。三周前一次网站改版把某个字段变成了 null。一个提取器的错误，如果不重新付费抓取所有数据就无法修复。

本指南关注的是加载这一端：如何可靠地、大批量地将抓取 API 的输出导入 SQL 数据库，同时不失去解释或重放所存储内容的能力。示例使用 PostgreSQL 和 Shifter Web Scraping API，其模式适用于任何 SQL 数据库。

从你实际得到的响应出发

使用 Shifter Web Scraping API，你可以在两种响应格式之间选择。

原始 HTML，由你自行解析。或者结构化 JSON，通过传入 extract_rules 将 CSS 选择器映射到字段：

curl "https://scrape.shifter.io/v1?api_key=YOUR_API_KEY&url=https://shop.example.com/p/42&render_js=1&extract_rules=%7B%22title%22%3A%7B%22selector%22%3A%22h1%22%2C%22output%22%3A%22text%22%7D%2C%22price%22%3A%7B%22selector%22%3A%22.price%22%2C%22output%22%3A%22text%22%7D%7D"

# {"title": "Example Product", "price": "$19.99"}

这种输出格式有两个特性会影响你的架构设计。一个选择器没有匹配到任何内容时，返回的是 null 而不是请求失败，因此一个缺失的元素和一个失效的选择器在响应中看起来是一样的。而且文本输出是展示用字符串，所以价格、评分和日期到达时是为人类格式化的，而不是为数据库定型的。规则语法，包括结果页的列表提取，见 提取规则文档。

三个层次，而非一张表

能在生产环境中存活下来的设计，会把你收到的内容和你得出的结论分开。

层次	内容	存在的原因
原始落地层	每一个成功的响应，按接收时的原样，附带抓取元数据	无需重新抓取即可重放提取过程
类型化观测层	已解析、已定型、已验证的值，附带解析状态	分析师和应用程序查询的对象
当前状态层	每个实体的最新值，由观测数据派生而来	供产品和仪表盘快速读取

原始层是团队最容易跳过、也最容易后悔的部分。积分花在了成功的请求上，所以如果解析中的一个错误只能通过重新抓取来修复，那整次抓取的成本就要付两遍。先落地响应，再解析，这样解析器的修复就变成了一次重放查询。

落地表

CREATE TABLE scrape_raw (
  job_id            text        PRIMARY KEY,
  source_url        text        NOT NULL,
  market            text        NOT NULL,
  fetched_at        timestamptz NOT NULL,
  http_status       smallint    NOT NULL,
  body              jsonb       NOT NULL,
  body_hash         text        NOT NULL,
  extractor_version text        NOT NULL
);

CREATE INDEX scrape_raw_url_time ON scrape_raw (source_url, fetched_at DESC);

其中有几个刻意的选择。

job_id 是幂等键，在请求之前根据 URL 和调度窗口计算得出，因此同一个逻辑任务的重试会落到同一个键上，而不会产生第二行记录。extractor_version 记录的是哪一套提取规则产生了该 body，这让你之后可以区分是网站发生了变化，还是规则发生了变化。market 记录观测发生的地点，因为同一个 URL 在不同国家可能返回不同的内容。而 API 密钥永远不会存储在你持久化的任何请求元数据中，因为数据库里的凭证意味着每一份备份里都存在凭证。

幂等加载

import hashlib
import json
import os

import psycopg
import requests
from psycopg.types.json import Jsonb

API = "https://scrape.shifter.io/v1"
RULES = {
    "title": {"selector": "h1", "output": "text"},
    "price": {"selector": ".price", "output": "text"},
}
EXTRACTOR_VERSION = "product-v3"

INSERT_RAW = """
INSERT INTO scrape_raw
  (job_id, source_url, market, fetched_at, http_status, body, body_hash, extractor_version)
VALUES (%s, %s, %s, now(), %s, %s, %s, %s)
ON CONFLICT (job_id) DO NOTHING
"""


def job_id(url: str, market: str, window: str) -> str:
    return hashlib.sha256(f"{url}|{market}|{window}".encode()).hexdigest()


def fetch(url: str, market: str) -> requests.Response:
    params = {
        "api_key": os.environ["SHIFTER_API_KEY"],
        "url": url,
        "render_js": 1,
        "country": market,
        "extract_rules": json.dumps(RULES),
    }
    return requests.get(API, params=params, timeout=90)


def land(conn: psycopg.Connection, url: str, market: str, window: str) -> None:
    resp = fetch(url, market)
    if resp.status_code != 200:
        raise RuntimeError(f"{resp.status_code} for {url}")
    body = resp.text
    with conn.cursor() as cur:
        cur.execute(
            INSERT_RAW,
            (
                job_id(url, market, window),
                url,
                market,
                resp.status_code,
                Jsonb(json.loads(body)),
                hashlib.sha256(body.encode()).hexdigest(),
                EXTRACTOR_VERSION,
            ),
        )

ON CONFLICT (job_id) DO NOTHING 是使这次加载可以安全重试的关键。无论失败原因是网络、你的工作进程，还是数据库，重新运行该任务都不会产生重复数据。对于 MySQL，等效的做法是配合唯一键使用 INSERT IGNORE 或 ON DUPLICATE KEY UPDATE。

吞吐量：将抓取与加载解耦

抓取端有一个由你套餐的并发上限设定的硬性天花板，超过该上限的请求会返回 429。数据库也有自己的天花板,而团队通常最先撞到的天花板，是为每一个抓取的页面都开一个连接和一个事务。

在两者之间放一个队列。抓取工作进程按照你的并发上限来设定规模，将响应写入队列。少量的加载工作进程分批消费队列。对于稳定的流量,psycopg 的 executemany 按几百行一批就足够了。对于大规模的回填,将批次 COPY 进一个未记录日志的暂存表，然后一条语句合并进去：

INSERT INTO scrape_raw
SELECT * FROM scrape_raw_staging
ON CONFLICT (job_id) DO NOTHING;

这将成千上万次往返合并为一次,同时保持幂等性保证不变。

对于长时间的渲染,API 可以异步交付：传入 webhook=<URL>，响应就绪时会被推送到你的端点。同样要让接收端在 job_id 上是幂等的,因为任何一方都可能对 HTTP 传输发起重试。

将 API 错误映射到管道行为

状态码并非都值得重试,如果加载器一视同仁地处理所有状态码,要么会对一个损坏的配置反复发起请求,要么会放弃本可恢复的临时故障。完整的表格见 错误与限制。

状态码	管道行为
408、422、500	使用指数退避进行重试
429	退避并降低工作进程并发数
400、401、403	配置错误：发送到死信队列并告警,永不重试
509	积分耗尽：停止抓取阶段并告警,重试无济于事

失败的请求以及目标返回的 4xx 或 5xx 响应不会被计费,而且 API 在返回之前已经对临时性故障自动重试最多三次,所以你自己的重试消耗的是时间而非积分。但它们仍然消耗时间,这正是退避机制重要的原因。

为观测数据定型

这是展示字符串变成数据的地方,也是大多数无声错误产生的地方。

CREATE TABLE price_observation (
  source_url    text          NOT NULL,
  market        text          NOT NULL,
  observed_at   timestamptz   NOT NULL,
  price_amount  numeric(12,2),
  currency      char(3),
  raw_price     text,
  parse_status  text          NOT NULL,
  job_id        text          NOT NULL REFERENCES scrape_raw (job_id),
  PRIMARY KEY (source_url, market, observed_at)
);

三条规则可以保持数据的可靠性。

将原始字符串与解析后的值放在一起。 raw_price 让你能够审查一个可疑的数字,而无需重新抓取。

按市场解析,而非全局解析。 "1.299,00" 和 "1,299.00" 在不同的书写惯例下是同一个价格,而货币符号并不等同于货币：$ 根据商店所在地不同,可能是美元、加元或澳元。要结合符号与市场共同解析出 ISO 代码。

记录一个值为何是 null。 parse_status 为 missing、unparseable 或 ok,可以区分”页面上没有价格”和”我们的解析器失败了”,而 API 返回的 null 本身无法告诉你这一点。

对于当前状态,应当派生而非维护。在 PostgreSQL 中：

CREATE VIEW price_current AS
SELECT DISTINCT ON (source_url, market) *
FROM price_observation
WHERE parse_status = 'ok'
ORDER BY source_url, market, observed_at DESC;

一个派生视图不可能与它所汇总的历史数据产生偏差。

在用户发现之前检测到架构漂移

网站会改变它们的标记结构,而一个改变的选择器并不会抛出错误。它会返回 null,请求成功,一个积分被消耗,而这行数据落地时看起来是有效的。

防御手段是按字段、按来源、按提取器版本进行 null 率监控。计算每个字段在滚动窗口内 missing 解析状态所占的比例,当它相对基线发生急剧变化时发出告警。一个价格字段在一夜之间从 2% 缺失变成 60% 缺失,说明是一次改版,而当天就发现它,和修复一个选择器与三周不可用的历史数据之间的区别就在于此。

修复之后,提升 extractor_version,并用新的解析器重放受影响的原始行。这就是落地原始响应所带来的回报。

保留策略与分区

原始落地表增长最快,却被读取得最少。按抓取日期对它们进行分区,保留时间要足以覆盖你实际可能需要的重放窗口,然后丢弃旧分区,而不是逐行删除。观测表是历史记录,通常应该获得更长的保留期,同样按日期分区。

需要监控的内容

指标	能发现的问题
成功抓取数与已落地行数的对比	加载器在 API 与数据库之间的数据丢失
job_id 重复冲突	重试风暴或调度重叠
每个字段和提取器版本的 null 率	标记结构变化和失效的选择器
队列深度和加载延迟	加载器落后于抓取阶段
消耗的积分与成功解析为 ok 的行数对比	花在无法使用的响应上的钱

最后这个指标才是真正重要的成本视角：每个可用行消耗的积分,而不是每个请求消耗的积分。API 本身的使用量和错误率可以在面板中的 Web Scraping API 下查看。

常见问题

我应该在原始层存储 HTML 还是提取后的 JSON？

提取后的 JSON 体积小得多,通常也足够使用。只有在你预期会经常更改提取逻辑的来源上才存储 HTML,并为该表设置较短的保留期。

JSONB 是否足够直接用于查询？

用于探索性分析,是足够的。而对于产品或仪表盘所依赖的任何内容,应将字段提升为类型化的列,让数据库能够强制类型约束并使用普通索引。

如何避免为未发生变化的页面付费？

每一次成功的请求都会消耗一个积分,所以节省成本要靠减少抓取,而不是减少写入。使用列表页或站点地图这类低成本信号,来决定哪些详情页确实需要抓取。

这套方法同样适用于 Amazon API 吗？

适用。它的响应已经是结构化 JSON,所以提取规则可以省去,但价格仍然以展示字符串的形式到达,落地、定型和漂移检测的模式不变地适用。

结论

抓取 API 解决的是采集问题。你的数据库设计决定了你所采集的数据是否值得信赖。用幂等键和提取器版本落地每一个成功的响应,按市场对值进行定型同时保留原始字符串,记录一个值为何是 null,派生当前状态而不是维护它,并按字段监控 null 率,从而让标记结构的变化在当天就浮出水面。

具体针对 Amazon 的方案对比见 面向 Amazon 监控的最佳网络抓取 API,同一套管道在房地产领域的版本见 房地产公司如何使用网络抓取 API。产品页面见 Web Scraping API,套餐信息见 定价页面。