网络爬取数据存入 SQL用 Python 存储与分析

网络爬取帮你获得数据，但存储和分析同样重要。一个松散 JSON 或 CSV 文件的文件夹，对于一次性拉取来说还好，但一旦你想要追踪价格随时间的变化、比较不同来源的记录，或者回答一个真实问题，这堆扁平文件就成了瓶颈。SQL 数据库为爬取数据提供了一个结构化的归宿，你可以在不每次都编写一次性解析代码的情况下查询、过滤和聚合。

本指南用 Python 和 SQL 从头到尾构建完整的路径。你通过 Crawling API 从一个中立示例页面爬取少量产品风格的记录，设计 SQL 架构，插入解析后的行，并运行分析查询来提取洞察。整个过程限定在公开的、说明性的数据和占位符 URL 上，所以你可以安全地跟着做，之后再换成你自己的目标。

你将构建什么

一个可运行的 Python 流程：通过 Crawling API 获取渲染后的列表页，用 BeautifulSoup 解析每个产品卡片，将记录写入 SQLite 表，然后查询该表以获得价格区间、平均值和类别计数。每条爬取的记录包含以下字段：

• 名称 列表卡片上显示的产品标题。
• 价格 标价，存储为数字，以便比较和聚合。
• 类别 商品所属的分区或产品类型。
• URL 指向产品自有页面的链接。
• 爬取时间 数据库自动填写的时间戳，让你可以追踪每行数据的收集时间。

为什么将爬取数据存储在 SQL 中

数据落地的位置决定了它之后有多大用处。文件可以用，直到你需要提出一个跨行的问题：哪些商品价格低于某个阈值，每个类别的平均价格是多少，本周的拉取与上周相比如何。SQL 正是为此而生的，关系型数据库能给你几个文件夹做不到的事情。

• 结构化存储。表和列强制执行一致的格式，确保每条记录都有相同类型的相同字段。
• 高效查询。单个 SELECT 语句无需任何手动循环就能过滤、排序和聚合数千行。
• 数据完整性。主键、类型和约束保持数据一致，并在入库时捕获格式错误的行。
• 可扩展性。无论你存储数千还是数百万条记录，SQL 引擎都能处理增长，并在正确的索引下保持速度。
• 可重复分析。查询是可重用的。一旦你写了一个有用的查询，就可以在明天的数据上再次运行，无需重建任何东西。

如果你在更广泛地权衡存储格式，JSON 与 CSV 比较涵盖了什么时候平面文件仍然是正确选择，数据建模指南对架构设计的深入程度超出了我们这里的篇幅。

为什么普通请求在现代列表页上会失败

在存储工作有意义之前，你需要有干净的记录可以存储，而这正是朴素的 HTTP 请求容易出问题的地方。许多现代列表页在浏览器中通过 JavaScript 渲染其内容：第一个 HTML 响应是一个薄薄的壳，产品卡片只有在页面脚本运行之后才会出现。从那个初始响应拉取数据，你捕获到的是列表的片段，或者什么都没有。

第二个问题是访问。发布商业上有价值的列表的网站会监控自动化流量，并封锁来自数据中心 IP 或任何看起来不像真实浏览器的请求。因此，可靠的爬虫需要在一次请求中做两件事：渲染页面的浏览器，以及网站认为是真实访客的 IP。你可以用无头浏览器和轮换住宅代理池自己构建，但保持那个技术栈健康运行是大部分工作所在。Crawling API 将两者合并为一次调用，并返回可以解析的完整 HTML。对于解析方面，Python 网络爬取指南和结构化爬取数据指南是有用的参考。

前提条件

在你写任何代码之前，应该先准备好几件事。都不需要太长时间。

基本的 Python 和 SQL 知识。你应该能够运行 Python 脚本并读懂一条 SELECT 语句。你不需要成为数据库管理员；本指南使用 SQLite，它随 Python 附带，所以没有服务器需要安装。

Python 3.8 或更高版本。用 python --version 确认你的版本。如果没有，请从 python.org 安装，并确保 Python 在你的 PATH 中。

Crawlbase 账号和令牌。注册后，打开控制台并复制你的令牌。Crawlbase 提供 1,000 个免费请求供开始使用，足以完成本指南的演练。像对待密码一样对待令牌，不要将其提交到版本控制中。如果你的目标在客户端渲染内容，使用 JavaScript（JS）令牌，这样页面在你收到之前就会被渲染。

搭建项目

创建一个虚拟环境以隔离依赖，然后安装爬虫需要的两个库。

python --version

python -m venv scraper_env
source scraper_env/bin/activate

pip install crawlbase beautifulsoup4

在 Windows 上，使用 scraper_env\Scripts\activate 代替 source 行来激活环境。crawlbase 包是 Crawling API 的官方客户端，beautifulsoup4 解析返回的 HTML，让你可以提取单独的字段。sqlite3 和 json 都随 Python 标准库附带，所以存储步骤不需要再安装任何东西。

第一步：获取渲染后的页面

从获取一个完整的页面开始。导入 CrawlingAPI 类，用你的令牌初始化它，然后请求列表 URL。我们使用占位符 https://example.com/products；将其指向任何你有权收集的公开列表页。在解析之前检查 Crawlbase 的 pc_status 可以让失败显眼而非悄无声息。

from crawlbase import CrawlingAPI

api = CrawlingAPI({"token": "YOUR_CRAWLBASE_TOKEN"})

LISTING_URL = "https://example.com/products"

def crawl(page_url):
    response = api.get(page_url)
    if response["headers"]["pc_status"] == "200":
        return response["body"].decode("utf-8")
    print(f"Request failed: {response['headers']['pc_status']}")
    return None

if __name__ == "__main__":
    html = crawl(LISTING_URL)
    print(html[:500] if html else "No HTML returned")

运行这段代码，你应该能看到打印出来的真实页面标记，这在你写任何选择器之前就确认了请求是有效的。如果你的目标在客户端渲染卡片，改用 JS 令牌，并将 {"ajax_wait": "true", "page_wait": 5000} 作为第二个参数传给 api.get，让动态内容有时间加载。

第二步：解析要存储的记录

有了完整的 HTML，将其加载到 BeautifulSoup 中，从每个产品卡片提取一条记录。下面的选择器假设一个带有标题、价格、类别标签和链接的卡片布局，这是列表页的常见形式。每次查找都有保护，使缺失字段返回安全默认值，而不是让整个运行崩溃。存储中最重要的细节是价格：将其作为文本爬取，然后去除货币符号和逗号，使其作为可以比较和平均的数字进入数据库。

import re
from bs4 import BeautifulSoup

def parse_price(text):
    if not text:
        return None
    cleaned = re.sub(r"[^0-9.]", "", text)
    return float(cleaned) if cleaned else None

def parse_products(html):
    soup = BeautifulSoup(html, "html.parser")
    records = []
    for card in soup.select("div.product-card"):
        name = card.select_one("h2.product-title")
        price = card.select_one("span.price")
        category = card.select_one("span.category")
        link = card.select_one("a.product-link")
        records.append({
            "name": name.get_text(strip=True) if name else None,
            "price": parse_price(price.get_text() if price else None),
            "category": category.get_text(strip=True) if category else "Uncategorized",
            "url": link["href"] if link and link.get("href") else None,
        })
    return records

parse_price 去除所有不是数字或小数点的字符，所以 "$1,299.00" 变成浮点数 1299.0。这个单一步骤使后续的价格查询成为可能：文本按字母排序，无法求平均值，但数字两者都能正确处理。将选择器（product-card、product-title、price、category、product-link）调整为你真实目标的标记；其余流程保持不变。

第三步：设计 SQL 架构

现在设计保存这些记录的表。好的架构反映你的记录形状，为每列选择正确的类型，并添加主键和数据库自动填写的时间戳。注意以下选择：price 是 REAL（数字，而非文本），因此可以比较和聚合，scraped_at 默认为当前时间，使每一行都记录其收集时间。

CREATE TABLE IF NOT EXISTS products (
    id        INTEGER PRIMARY KEY AUTOINCREMENT,
    name      TEXT NOT NULL,
    price     REAL,
    category  TEXT,
    url       TEXT UNIQUE,
    scraped_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Speed up the price filters and category rollups you will run later
CREATE INDEX IF NOT EXISTS idx_price ON products(price);
CREATE INDEX IF NOT EXISTS idx_category ON products(category);

有两个设计选择值得指出。url 上的 UNIQUE 约束阻止同一产品被插入两次，这在爬虫重新运行时很重要，你不希望昨天的行在今天被重复。索引随着表的增长保持价格和类别查询的速度。在 MySQL 或 PostgreSQL 上，架构读起来几乎相同：用 AUTO_INCREMENT 或 SERIAL 代替 AUTOINCREMENT，对于金钱字段 DECIMAL(10,2) 是比 REAL 更严格的选择。数据建模指南涵盖了每种类型何时是正确选择。

第四步：插入已解析的记录

设计好表之后，从 Python 连接，如果需要就创建表，然后插入解析后的记录。始终使用参数化查询：? 占位符让驱动程序处理转义，这既防止了格式错误的文本破坏插入语句，也关闭了注入漏洞。executemany 在一次批处理调用中写入所有记录，当你有数百行时比单个插入的循环快得多。

import sqlite3

SCHEMA = """
CREATE TABLE IF NOT EXISTS products (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    name TEXT NOT NULL,
    price REAL,
    category TEXT,
    url TEXT UNIQUE,
    scraped_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
)
"""

def store_records(records, db_path="scraped_data.db"):
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    cursor.execute(SCHEMA)

    rows = [
        (r["name"], r["price"], r["category"], r["url"])
        for r in records if r["name"]
    ]
    cursor.executemany(
        "INSERT OR IGNORE INTO products (name, price, category, url) "
        "VALUES (?, ?, ?, ?)",
        rows,
    )

    conn.commit()
    inserted = cursor.rowcount
    conn.close()
    return inserted

INSERT OR IGNORE 与第三步中 url 上的 UNIQUE 约束配合：URL 已存在的行会被静默跳过，而不是抛出错误，所以重新运行爬虫会补充新列表，而不会重复旧列表。批处理后提交一次保持写入速度。对于 MySQL，相同的代码形式适用，用 mysql.connector 代替 sqlite3，用 %s 占位符代替 ?；下游的一切都是相同的。

第五步：用 SQL 查询分析数据

数据现在处于可查询的形式，这正是使用 SQL 的全部意义所在。以下是你会经常用到的三种分析：过滤和排序、聚合汇总，以及按类别分组获得洞察。每个都是普通的 SELECT，你可以从 SQLite shell、数据库 GUI 或 Python 中运行。

首先，过滤和排序。这会提取价格低于某个阈值的所有产品，最便宜的排在前面：

SELECT name, price, category
FROM products
WHERE price < 500
ORDER BY price ASC;

然后，将整个表聚合为一行汇总。COUNT、AVG、MIN 和 MAX 等函数将多行折叠为你真正想要报告的数字：

SELECT
    COUNT(*) AS total_products,
    ROUND(AVG(price), 2) AS average_price,
    MIN(price) AS cheapest,
    MAX(price) AS most_expensive
FROM products;

最后，按类别分组以查看目录的构成。GROUP BY 为每个类别运行一次聚合，HAVING 在事后过滤这些分组：

SELECT
    category,
    COUNT(*) AS items,
    ROUND(AVG(price), 2) AS avg_price
FROM products
GROUP BY category
HAVING COUNT(*) > 1
ORDER BY avg_price DESC;

要从 Python 而不是 shell 运行这些查询，打开相同的连接并读取行。这个片段运行按类别汇总并打印每个分组：

def category_summary(db_path="scraped_data.db"):
    conn = sqlite3.connect(db_path)
    cursor = conn.cursor()
    cursor.execute("""
        SELECT category, COUNT(*), ROUND(AVG(price), 2)
        FROM products
        GROUP BY category
        ORDER BY 3 DESC
    """)
    for category, count, avg_price in cursor.fetchall():
        print(f"{category}: {count} items, avg {avg_price}")
    conn.close()

如果你的分析超出了普通 SQL 的舒适范围，将表拉入 DataFrame 然后继续；pandas 分析指南正好从这个点接着往下讲。

将其整合为一个流程

这四个函数链接为一个可运行的脚本。爬取、解析、存储，然后分析：

def main():
    html = crawl(LISTING_URL)
    if not html:
        return
    records = parse_products(html)
    inserted = store_records(records)
    print(f"Parsed {len(records)} records, inserted {inserted} new rows")
    category_summary()

if __name__ == "__main__":
    main()

每次运行获取页面，由于 INSERT OR IGNORE 只写入新行，并打印按类别汇总。在 cron 任务上调度它，表就成为一个你可以随时间查询的增长历史，这正是 SQL 相比一堆带日期的 CSV 文件所能体现价值的地方。

输出的样子

一条存储的记录带有结构化字段加上数据库分配的 id 和时间戳。从中查询几行返回的内容类似这样：

[
  {
    "id": 1,
    "name": "Wireless Headphones",
    "price": 129.99,
    "category": "Audio",
    "url": "https://example.com/products/wireless-headphones",
    "scraped_at": "2026-06-11 09:42:18"
  },
  {
    "id": 2,
    "name": "Mechanical Keyboard",
    "price": 89.50,
    "category": "Accessories",
    "url": "https://example.com/products/mechanical-keyboard",
    "scraped_at": "2026-06-11 09:42:18"
  }
]

对同一个表运行类别汇总查询，会打印出紧凑的汇总，每个类别一行，包含计数和平均价格，这正是一堆原始文件在没有额外代码的情况下无法给你的洞察。

扩展到单个页面之外

上面的流程处理一个页面。真实的数据集跨越许多页面，一些调整可以将这个演示升级为生产就绪的形式。

• 分页。大多数列表页会暴露页面参数或下一个链接。遍历页面，为每个页面调用 crawl 和 parse_products，并将合并的记录一次性传给 store_records。
• 批量插入。executemany 和每批一次的 commit 已经最小化了往返次数。对于非常大的任务，每隔几千行提交一次，而不是全部一起，这样失败就不会丢失整个运行。
• 为你查询的列建立索引。第三步的两个索引覆盖了价格和类别过滤。在你频繁过滤或排序的任何其他列上添加索引。
• 大量访问时使用异步。对于数千个页面，异步 Crawler 将请求排队并将结果发送到 webhook，这样你就不必在爬取时保持连接打开。与批量插入配合，数据库可以跟上。

如果你在设计更大的收集-解析-存储-分析系统而不是单个脚本，数据管道架构指南涵盖了这些阶段如何在规模上组合在一起。

负责任地爬取

将收集限制在合理范围内。只爬取不在登录后面的公开数据，阅读并遵守网站的服务条款和 robots.txt，并将请求速率保持在合理范围内，以免给目标服务器造成压力。当数据涉及可识别个人时，GDPR 和 CCPA 等隐私规则适用，因此除非你有合法依据和明确需要，否则应避免个人数据。这里的示例使用占位符 URL 和说明性产品字段，正是为了远离这些顾虑；在将爬虫指向真实网站时也适用同样的判断。

常见问题

为什么将爬取数据存储在 SQL 中而不是 CSV 或 JSON 文件？

文件对于单次拉取很好，但它们使跨行问题变得痛苦。SQL 数据库强制执行一致的结构，让你用一个查询过滤、排序和聚合，并随索引扩展到数百万行。它还能干净地处理重复运行：UNIQUE 约束阻止重复，所以你可以随时间向同一个表追加，而不是管理带日期的文件。

我应该使用哪个数据库，SQLite、MySQL 还是 PostgreSQL？

SQLite 非常适合小到中型项目和本地工作，因为它随 Python 附带且不需要服务器，这就是本指南使用它的原因。MySQL 适用于 Web 应用程序和较大的共享数据集，PostgreSQL 在复杂查询和分析方面很强。当你在它们之间切换时，这里的 Python 流程只改变连接和占位符语法；架构和查询几乎保持相同。

如何避免两次插入同一条记录？

在标识记录的列上放置 UNIQUE 约束，例如产品 URL，然后用 INSERT OR IGNORE（或 PostgreSQL 上的 ON CONFLICT）插入。数据库会跳过其键已存在的任何行，所以重新运行爬虫只会添加新列表，而不是重复旧列表。

为什么在存储之前将价格转换为数字？

因为以文本形式保存的价格（如 "$1,299.00"）无法正确比较或求平均值：它按字母排序，并破坏聚合函数。去除符号和逗号以存储数值 price，意味着 WHERE price < 500、AVG(price) 和 ORDER BY price 都能按预期工作。

我需要 JavaScript 令牌吗？

只有当你的目标在客户端渲染其内容时。普通令牌返回静态 HTML，对于服务端渲染的页面来说已经足够。如果列表在页面脚本运行后才填充，使用 JS 令牌并添加 ajax_wait 和 page_wait 选项，这样动态内容在 HTML 被捕获之前就能加载完毕。

如何将其扩展到多个页面而不使数据库过载？

对爬取进行分页，并将合并的记录传给批量的 executemany 插入，每隔几千行提交一次而不是每次一行。对于大型爬取，异步 Crawler 将请求排队并将结果发送到 webhook，这样你就不必保持连接打开，对你查询的列建立索引使随着表的增长读取保持快速。

HR

Hassan Rehan

软件工程师 · Crawlbase

Crawlbase 软件工程师，撰写关于轮换代理、抓取，以及把代理接入真实代码的实战细节的实操指南。