如何将抓取数据存储到云端用 Python 实现持久存储

抓取页面只是工作的一半。你提取的数据必须存放在某个持久的地方，供团队成员和分析工具访问，并且不会因笔记本电脑崩溃而丢失。本地 CSV 文件用于一次性任务尚可，但一旦抓取变成定期流程，单块硬盘就成了隐患：容量会耗尽，机器间传输繁琐，一次磁盘故障就可能让你损失无法挽回的数据。

本指南将向你展示如何用 Python 端到端地将抓取数据存储到云端。你将构建一个小型的可运行流程：通过 Crawling API 抓取页面，将结果整理为干净的记录，然后写入持久的云端目标, 使用 S3 风格存储桶的对象存储，以及托管关系型数据库。所有示例均使用中性示例 URL 和环境变量占位符作为凭据，方便你在不改变代码结构的情况下适配自己的目标和云服务商。

你将构建什么

一个 Python 脚本，通过 Crawling API 从公开示例列表页抓取少量数据集，将每行数据规范化为带类型的记录，然后以两种方式将这些记录发送到云端。你可以只保留其中一条路径，也可以两条都保留。各组成部分如下：

• 抓取：通过 Crawling API 获取已渲染的页面，返回完整 HTML。
• 转换：将原始 HTML 转换为字段名和类型一致的结构化记录列表。
• 对象存储：将 JSON Lines 文件上传到 S3 风格的存储桶，用于廉价持久的归档。
• 托管数据库：将相同的记录插入 Postgres 表，以便查询和关联。
• Crawlbase Cloud Storage：一个可选的单参数路径，在服务端保存原始抓取响应。

为何将抓取数据存储在云端

本地存储在未出问题之前很方便。当抓取规模从数百行增长为定期向仪表盘供数据的常规任务时，三个问题会同时出现。容量成为持续成本：你需要购买磁盘来保证备份安全，还要花时间管理它们。访问变得尴尬：锁在一台机器上的数据很难与团队共享，或馈送给在其他地方运行的工具。持久性也很脆弱：电源问题、固件损坏和简单的人为失误都可能让一块磁盘瘫痪，连同所有未经备份的数据一起消失。

云存储解决了这三个问题。对象存储和托管数据库天生具备冗余能力，数据跨多个地点复制，而不是停在一块硬盘上。它们无需你自己配置硬件即可扩展，只要有凭据就可从任何地方访问，还将备份和持久性职责移交给了云服务商。对于抓取流程而言，这意味着你可以从数据落地的那一刻起就将其视为持久资产，而不是需要在本地磁盘上费心照料的东西。

两种目标，以及各自的适用场景

本教程写入两种云存储，因为抓取数据通常都需要这两种。对象存储（S3 兼容存储桶）是原始和归档数据的理想归宿：按 GB 收费低廉，对文件格式无要求，非常适合保存未经处理的抓取输出，以便日后重新处理。托管关系型数据库（此处为 Postgres）是结构化可查询副本的理想归宿，一致的列和类型让你可以用 SQL 进行过滤、聚合和关联。常见模式是同时写入两者，下面的代码正是这样做的。更深入的对比请参阅 云存储与本地存储 以及 云存储的优势。

前提条件

在写代码之前，需要先准备好几样东西。都不复杂。

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

Crawlbase 账户和 token。注册后打开仪表盘，复制你的 token。Crawlbase 最多提供 20,000 次免费请求作为起点，足够完成本指南的学习。请像对待密码一样保护 token，不要将其纳入版本控制。如果目标页面的内容在客户端渲染，请使用 JavaScript token；对于静态页面，普通 token 即可。

云凭据。对象存储路径需要 S3 风格的存储桶和一对访问密钥。数据库路径需要托管 Postgres 实例的连接字符串。两者均通过以下代码中的环境变量提供，绝不硬编码。

熟悉 Python 和基本抓取知识。如果你对解析部分还不熟悉，BeautifulSoup 指南和 用 Python 抓取网站 的教程是很好的参考。

搭建项目

创建虚拟环境以隔离依赖，然后安装流程所需的库。

python --version

python -m venv cloud_env
source cloud_env/bin/activate

pip install crawlbase beautifulsoup4 boto3 psycopg2-binary

在 Windows 上，用 cloud_env\Scripts\activate 替换 source 那行来激活环境。四个依赖各司其职：crawlbase 是 Crawling API 的官方客户端，beautifulsoup4 用于解析返回的 HTML，boto3 负责与 S3 风格对象存储通信，psycopg2-binary 连接 Postgres。json 模块随标准库附带，因此归档格式不需要额外安装。

第 1 步：通过 Crawling API 抓取页面

首先获取已渲染的页面。导入 CrawlingAPI 类，用你的 token 初始化它，然后请求目标 URL。在解析前检查 Crawlbase 的 cb_status（legacy pc_status） 可以让失败变得显而易见，而不是悄然无声。我们在这里使用中性示例列表页；在适配流程时请替换为你自己的 URL。

from crawlbase import CrawlingAPI

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

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

if __name__ == "__main__":
    page_url = "https://example.com/products"
    html = crawl(page_url)
    print(html[:500] if html else "No HTML returned")

用 python cloud_pipeline.py 运行此脚本，你应该会看到打印出真实的页面标记，在编写任何选择器之前先确认抓取正常工作。如果目标在客户端填充内容，请用你的 JavaScript token 初始化客户端，并向 api.get 传入 {"ajax_wait": "true", "page_wait": 5000}，让 API 先渲染页面。对于 JS 重度目标，用 Python 抓取 JavaScript 页面指南涵盖了相关细节。

第 2 步：将 HTML 转换为结构化记录

原始 HTML 不适合直接存入数据库。转换步骤将其变为字段名和类型一致的字典列表，使每条记录具有相同的结构。将 HTML 加载到 BeautifulSoup，遍历页面上的每个条目，提取你需要的字段。这里的选择器仅供示意；请替换为与你目标匹配的选择器。

from bs4 import BeautifulSoup

def text_of(node, selector):
    el = node.select_one(selector)
    return el.get_text(strip=True) if el else None

def to_price(raw):
    if not raw:
        return None
    digits = raw.replace("$", "").replace(",", "").strip()
    return float(digits) if digits.replace(".", "").isdigit() else None

def transform(html):
    soup = BeautifulSoup(html, "html.parser")
    records = []
    for card in soup.select("div.product"):
        records.append({
            "name": text_of(card, "h2.title"),
            "price": to_price(text_of(card, "span.price")),
            "sku": text_of(card, "span.sku"),
            "in_stock": text_of(card, "span.stock") == "In stock",
        })
    return [r for r in records if r["name"]]

三个小辅助函数让记录保持整洁。text_of 返回元素的去空白文本，若元素缺失则返回 None，这样某张卡片的空缺不会导致循环崩溃。to_price 去掉货币符号和千位分隔符并转换为浮点数，使数据库列可以是数字类型而非文本。最后的过滤器丢弃没有名称的行，这些行通常是布局残留而非真实条目。结果是一个带类型的记录列表，可以直接存储。关于如何更好地组织抓取数据，请参阅 结构化和清洗网络抓取数据。

第 3 步：上传到 S3 风格对象存储

第一个云端目标是对象存储。对象存储是原始或归档数据的天然归宿：廉价、持久，且对存入内容的格式无要求。我们将记录写成 JSON Lines 格式（每行一个 JSON 对象），易于追加，也便于后续流式读取。凭据来自环境变量，以确保源代码中不含任何敏感信息。

import os
import json
import boto3

def upload_to_s3(records, key):
    s3 = boto3.client(
        "s3",
        endpoint_url=os.environ.get("S3_ENDPOINT_URL"),
        aws_access_key_id=os.environ["S3_ACCESS_KEY"],
        aws_secret_access_key=os.environ["S3_SECRET_KEY"],
    )
    body = "\n".join(json.dumps(r) for r in records)
    s3.put_object(
        Bucket=os.environ["S3_BUCKET"],
        Key=key,
        Body=body.encode("utf-8"),
        ContentType="application/x-ndjson",
    )
    print(f"Uploaded {len(records)} records to s3://{os.environ['S3_BUCKET']}/{key}")

endpoint_url 参数使本方案不局限于 AWS：将其留空以使用 AWS S3，或将其指向任何 S3 兼容的服务商（例如自托管的 MinIO 实例或其他云的对象存储）。运行前设置好四个环境变量，例如 export S3_BUCKET=my-scrape-archive 以及对应的密钥。Key 是存储桶中的对象路径；像 scrapes/2026-06-11/products.jsonl 这样带日期戳的键可以让每次运行的归档相互独立，便于查找。

第 4 步：插入托管数据库

第二个目标是托管 Postgres 数据库，这里存放用于查询的结构化副本。下面的函数从单个环境变量打开连接，确保目标表存在，然后插入记录。使用参数化查询（%s 占位符）可以正确转义值，而不是将其拼接进 SQL。

import os
import psycopg2
from psycopg2.extras import execute_values

CREATE = """
CREATE TABLE IF NOT EXISTS products (
    id SERIAL PRIMARY KEY,
    name TEXT NOT NULL,
    price NUMERIC,
    sku TEXT,
    in_stock BOOLEAN,
    scraped_at TIMESTAMPTZ DEFAULT now()
)
"""

def save_to_db(records):
    conn = psycopg2.connect(os.environ["DATABASE_URL"])
    with conn, conn.cursor() as cur:
        cur.execute(CREATE)
        rows = [(r["name"], r["price"], r["sku"], r["in_stock"]) for r in records]
        execute_values(
            cur,
            "INSERT INTO products (name, price, sku, in_stock) VALUES %s",
            rows,
        )
    conn.close()
    print(f"Inserted {len(records)} rows into products")

将 DATABASE_URL 设置为托管 Postgres 的连接字符串，例如 postgresql://user:pass@host:5432/dbname，并将其保存在环境中而非文件里。CREATE TABLE IF NOT EXISTS 使函数可以安全地重复运行，scraped_at 列为每次加载打上时间戳以便追踪变更，而 execute_values 将插入操作批量合并为一次往返，而不是每行一次查询。数据入库后，你可以用纯 SQL 进行过滤和聚合，再将其拉入 pandas 进行分析。

第 5 步：组装完整流程

现在将各步骤串联成一个可运行的脚本：抓取、转换，然后将记录发送到两个目标。保留适合你工作流的存储调用；此处两者均已展示。

import os
import json
from datetime import date
from crawlbase import CrawlingAPI
from bs4 import BeautifulSoup

# crawl, transform, upload_to_s3 and save_to_db are defined above

def main():
    page_url = "https://example.com/products"
    html = crawl(page_url)
    if not html:
        print("Nothing scraped, stopping.")
        return

    records = transform(html)
    print(f"Parsed {len(records)} records")
    if not records:
        return

    key = f"scrapes/{date.today().isoformat()}/products.jsonl"
    upload_to_s3(records, key)
    save_to_db(records)

if __name__ == "__main__":
    main()

流程线性清晰，易于推理：获取页面，若抓取失败则提前退出，将 HTML 转换为记录，若无内容可存则再次退出，然后将记录归档到存储桶并加载到数据库。日期戳对象键让每次运行的归档相互独立，而数据库则通过时间戳累积每次加载。设置好环境变量后用 python cloud_pipeline.py 运行。

输出看起来是什么样的

对象存储路径写入一个 JSON Lines 文件，每条记录占一行，这就是落入存储桶的内容：

{"name": "Aluminium Tripod", "price": 129.99, "sku": "TRP-014", "in_stock": true}
{"name": "USB-C Hub", "price": 39.5, "sku": "HUB-203", "in_stock": false}
{"name": "Wireless Mouse", "price": 24.0, "sku": "MSE-088", "in_stock": true}

数据库路径将相同的记录存储为带类型的列，因此一个简单的查询就能确认加载结果并展示可供分析的结构：

SELECT name, price, in_stock FROM products WHERE in_stock = true ORDER BY price;

--      name       | price  | in_stock
-- ----------------+--------+----------
--  Wireless Mouse |  24.00 | t
--  Aluminium Tripod| 129.99 | t

两份副本就位后，你拥有了廉价持久的原始记录归档和一个可查询的结构化表，由同一次运行写入。

一个单参数快捷方式：Crawlbase Cloud Storage

如果你的目标只是在服务端保留每次抓取响应的副本，而不必先搭建自己的存储桶或数据库，Crawlbase Cloud Storage 提供了一个单参数路径。在 Crawling API 请求中添加 &store=true，响应副本就会自动保存到云端，你可以通过 API 或仪表盘对其进行搜索、检索或删除。

from crawlbase import CrawlingAPI

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

response = api.get("https://example.com/products", {"store": "true"})
# the response also includes a storage RID you can use to fetch it later
print(response["headers"].get("storage_url"))

每个保存的请求都有一个唯一标识符（RID），可用于查看或删除它。这是保留原始响应最快捷的方式，与 异步 Crawler 搭配使用效果尤佳，尤其是在发出大量请求并希望服务端处理存储的场景下。对于较大的结构化数据集，你仍然需要自己的数据库，但就原始响应归档而言，store 参数在简洁性上难以超越。

扩展流程

上面的流程只抓取一个页面。将其变成定期任务，主要关乎节奏和健壮性。几个习惯可以让更大规模的运行保持健康：

• 批量写入。积累记录后批量上传或插入，而不是逐行操作。execute_values 已经批量处理数据库插入；对象上传也采用同样的方式，每次运行写一个文件而非每条记录一个文件。
• 为键打日期戳。使用像 scrapes/2026-06-11/products.jsonl 这样带日期的对象键，使每次运行相互隔离，不会覆盖历史记录。数据库的 scraped_at 列在查询端起到同样的作用。
• 按计划运行。将脚本封装在 cron 任务或计划任务中，保持云端副本持续更新。由于表使用了 IF NOT EXISTS，且存储桶键带有日期，重复运行是安全的。
• 规模扩大时卸载抓取任务。对于大量页面，异步 Crawler 可将请求排队并通过 webhook 传递结果，适合高吞吐量场景，无需保持长连接。

负责任地抓取

请负责任地收集数据。只抓取公开数据，遵守每个网站的服务条款及其 robots.txt，保持合理的请求频率，不要给你所依赖的服务器造成压力。当你收集的数据涉及可识别个人身份的信息时，GDPR 和 CCPA 等隐私法律适用，因此除非你有合法依据和明确目的，否则应避免收集个人数据。将数据存储在云端不会改变上述要求：收集时的谨慎态度同样延伸到你保留数据的方式，而将个人数据保留超过必要时限只会增加风险。

常见问题

抓取数据应该存储在对象存储还是数据库中？

取决于你如何使用它。对象存储（S3 兼容存储桶）廉价持久，适合任何形状的原始或归档数据，是保存未处理抓取输出的理想归宿。托管关系型数据库适合你需要用 SQL 查询、过滤和关联的结构化副本。许多流程同时写入两者：将原始记录归档到存储桶，将清洗后的记录加载到数据库。

如何让云凭据不出现在代码中？

从环境变量或密钥管理器中读取，而不是硬编码。本指南中的代码从 os.environ 读取 S3 密钥和 Postgres 连接字符串，因此提交的文件中不含任何敏感信息。进入版本控制的密钥就是你必须轮换的密钥，所以请将其保存在环境中。

Crawlbase Cloud Storage 与上传到自己的存储桶有何区别？

Crawlbase Cloud Storage 是单参数路径：在 Crawling API 请求中添加 &store=true，原始响应就会在服务端保存，通过 RID 可检索，无需自己搭建任何基础设施。上传到自己的存储桶或数据库则让你完全掌控格式、模式、保留策略和存储位置，这对结构化数据集而言是必要的。两者互补：store 参数用于快速原始响应归档，自有存储用于经过处理的数据。

S3 代码可以与 AWS 以外的服务商一起使用吗？

可以。boto3 客户端有一个 endpoint_url 参数；将其留空以使用 AWS S3，或将其指向任何 S3 兼容的服务商，例如自托管的 MinIO 实例或其他云的对象存储。其余代码保持不变，这也是示例从环境变量读取 endpoint 的原因。

如何定期运行这个脚本以保持云端副本的最新状态？

将脚本封装在 cron 任务或计划任务中，以与数据变化频率匹配的节奏运行。流程可以安全重复：数据库表使用 CREATE TABLE IF NOT EXISTS，对象键带有日期戳所以运行之间不会相互覆盖，每条数据库行都带有 scraped_at 时间戳以便追踪变更。对于大量页面，将抓取任务交给异步 Crawler，这样任务就不会因单个连接而形成瓶颈。

将抓取的个人数据存储在云端安全吗？

首先将其视为法律和隐私问题。除非你有合法依据和明确目的，否则应避免收集与可识别个人相关的数据，因为 GDPR 和 CCPA 等隐私法律无论数据存储在哪里都适用。如果你确实持有个人数据，只存储必要的内容，保留时间不超过必要期限，并对访问进行安全控制。将个人数据保留超过必要时限只会增加风险而不会增加价值。

BA

Bilal Ahmed

软件工程师 · Crawlbase

软件工程师，在 Crawlbase 博客上撰写了一些阅读量最高的文章，涵盖网页抓取、代理与数据工具。