6.2 数据持久化：文件、SQLite 与 MySQL

6.2 数据持久化：文件、SQLite 与 MySQL

本节摘要：持久化有三档：框架内置的导出（JSON、CSV 等）零代码可用，SQLite 适合单机小体量，MySQL 适合正式项目与后续查询。本节给出三档的落地代码与选型矩阵，并讲清增量写入、唯一键防重、字段类型这些落地细节。

上一节管道把货洗好了，本节解决"存到哪"。选型的判断轴只有两条：数据规模与下游用途。先把三档的代码备好，再谈怎么选。

第一档：内置导出，零管道代码

框架自带 Item 导出器，运行时用 -o 参数即可产出多种格式，连管道都不用写：

scrapy crawl books -o books.json          # JSON，逐行产出
scrapy crawl books -o books.jsonl         # JSON Lines，流式追加友好
scrapy crawl books -o books.csv           # CSV，电子表格可直接打开
​

# books.jsonl 片段
{"title": "A Light in the Attic", "price": 51.77, "stock": true}
{"title": "Tipping the Velvet", "price": 23.19, "stock": true}
​

这一档适合探索期与一次性交付：零代码、即取即用。短板同样明确：无主键防重、无字段约束、不支撑并发追加——数据一多、任务一长，就该升档。导出格式的细节（编码、字段映射）可在配置里微调：FEED_EXPORT_ENCODING 指定编码避免中文转义。

第二档：SQLite，单机的轻量选择

SQLite 免服务、单文件、SQL 可查，是单机项目的甜点档：

import sqlite3
from itemadapter import ItemAdapter

class SqlitePipeline:
    def open_spider(self, spider):
        self.conn = sqlite3.connect("books.db")
        self.conn.execute(
            "CREATE TABLE IF NOT EXISTS book ("
            "url TEXT PRIMARY KEY, title TEXT, price REAL, stock INTEGER)"
        )

    def process_item(self, item, spider):
        ad = ItemAdapter(item)
        # url 作主键：INSERT OR IGNORE 天然防重，重复件静默跳过
        self.conn.execute(
            "INSERT OR IGNORE INTO book (url, title, price, stock) "
            "VALUES (?,?,?,?)",
            (ad["detail_url"], ad["title"], ad["price"], int(ad["stock"])),
        )
        self.conn.commit()
        return item

    def close_spider(self, spider):
        self.conn.close()
​

主键防重是本档的重点动作：把业务唯一键（detail_url）设为主键，配合 INSERT OR IGNORE，重复数据在库门口就被拦下。注意 SQLite 的写并发受限，多进程同时写同一文件会报锁错误——多任务并行时按任务拆文件，或直接升档。

第三档：MySQL，正式项目的默认档

正式项目、多人消费数据、需要与其他系统集成时，上 MySQL。6.1 节已给出批量落库管道，这里补齐建表与增量细节：

-- 建表：唯一键承担防重，字段类型一次定准
CREATE TABLE IF NOT EXISTS book (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    url VARCHAR(512) NOT NULL,
    title VARCHAR(256) NOT NULL,
    price DECIMAL(10,2) NULL,
    stock TINYINT DEFAULT 0,
    crawled_at DATETIME DEFAULT CURRENT_TIMESTAMP,
    UNIQUE KEY uk_url (url(255))
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
​

# 管道侧改写为 UPSERT：老数据更新、新数据插入
sql = (
    "INSERT INTO book (url, title, price, stock) VALUES (%s,%s,%s,%s) "
    "ON DUPLICATE KEY UPDATE title=VALUES(title), price=VALUES(price), "
    "stock=VALUES(stock)"
)
​

增量采集的核心是 UPSERT 语义：同一地址再抓到时更新而非报错。crawled_at 默认当前时间戳，为 5.3 节强调的"来源留痕"兜底。

案例展开：从导出到建库的一次完整迁移

背景：试跑期用 JSONL 导出攒了两周数据，需求升级为多端查询，决定迁到 MySQL。操作与踩坑实录。

第一步建表：按 6.1 的 Item 契约反推字段类型，price 用 DECIMAL 而不是 FLOAT（金额字段用浮点是埋雷），url 建 255 长度前缀唯一键（全串 512 超出索引限制）。第二步写导入脚本：逐行读 JSONL，按清洗管道的同一套口径转换后批量插入——注意是"同一套口径"，迁移脚本绕过清洗逻辑是脏数据混入的头号通道。第三步对账：源文件行数、插入行数、被唯一键拒掉的行数三数相核。

# 迁移对账的最小逻辑
import json

def migrate(jsonl_path, conn):
    total, dup, ok = 0, 0, 0
    for line in open(jsonl_path, encoding="utf-8"):
        total += 1
        row = json.loads(line)
        cur = conn.execute(sql, (row["url"], row["title"], row["price"], row["stock"]))
        if cur.rowcount == 0:
            dup += 1                    # 唯一键拒绝：库中已有
        else:
            ok += 1
    print(total, ok, dup)               # 三数对上才算迁移完成
​

结果与解读：两万一千行里对出四百行重复——来自导出期间两次重叠试跑，唯一键在迁移时就地拦下，没有流入下游。变式：数据要再进搜索引擎或报表系统时，同一套对账逻辑在管道落库层后再挂一层推送（6.5 节的下游集成），逐层对账是数据工程的通用纪律。

选型矩阵与落地清单

图13 存储选型矩阵：规模与用途两条轴

⚠️ 常见坑：CSV 与 Excel 打开的中文乱码，多半是导出编码没指定。给 FEED_EXPORT_ENCODING 明确赋值，或统一用 JSONL 交付，别让最后一步毁掉整条流水线的体面。

本节要点回顾

• 三档按需升档：快照用导出、单机用 SQLite、正式上 MySQL，迁移方向单向；
• 防重靠唯一键：导出档没这能力，所以它只配一次性任务；
• 增量用 UPSERT：老数据更新新数据插入，crawled_at 默认留痕；
• 编码问题前置解决：导出配置显式指定，别等打开乱码再排查。

数据落定了。下一节回答"一台机器不够怎么办"——什么时候真的需要分布式。