MongoDB 与半结构化数据

采集数据的字段是会长出来的：今天只要标题和链接，明天要加作者、标签列表、商品规格、SKU 价格区间。用 MySQL 意味着每加一个字段就要 ALTER TABLE，而字段在不同站点之间差异极大时，宽表里会堆满 NULL。MongoDB 的文档模型正适合这类半结构化数据，本章讲清它适合什么、最小可用 API、如何做幂等写入，以及什么时候不该用它。

MongoDB 适合什么、不适合什么

特征	适合 MongoDB	更适合 MySQL
字段结构	每个站点字段不同、经常新增	字段稳定、类型固定
嵌套结构	规格参数、标签数组、多价格档	需要拆成多表关联
迭代速度	加字段不用改表结构	变更要走 DDL 与发布流程
事务要求	单文档原子性够用	多表强事务、复杂关联查询
查询模式	按 `_id`/`url` 点查、按时间范围过滤	多条件组合、聚合报表、JOIN
数据量	亿级文档横向分片方便	千万级以内单库更省心

判断标准很简单：如果字段集合在不同数据源之间不一致，或者数据天然是嵌套的（一个商品多个 SKU），优先 Mongo；如果数据要频繁多表 JOIN 做统计，优先 MySQL。

最小可用 API

# mongo_store.py —— Python 3.10+，依赖 pymongo 4.x
import hashlib

from pymongo import ASCENDING, MongoClient, UpdateOne
from pymongo.errors import BulkWriteError


class MongoStore:
    def __init__(self, uri: str = "mongodb://127.0.0.1:27017", db: str = "spider"):
        self.client = MongoClient(uri, serverSelectionTimeoutMS=3000)
        self.col = self.client[db]["news"]

    def ensure_index(self):
        # 唯一索引是整个去重方案的基础，必须先建
        self.col.create_index([("url_md5", ASCENDING)], unique=True)
        # 复合索引：支撑「按站点 + 时间倒序」的常用查询
        self.col.create_index([("site", ASCENDING), ("pub_time", ASCENDING)])

    def upsert_many(self, rows: list[dict]) -> dict:
        """批量幂等写入：有则更新、无则插入，一次网络往返"""
        ops = []
        for row in rows:
            row = dict(row)
            row["url_md5"] = hashlib.md5(row["url"].encode("utf-8")).hexdigest()
            ops.append(UpdateOne({"url_md5": row["url_md5"]}, {"$set": row}, upsert=True))
        try:
            result = self.col.bulk_write(ops, ordered=False)  # 无序，可并行，单条失败不影响其他
            return {"upserted": result.upserted_count, "modified": result.modified_count}
        except BulkWriteError as exc:
            details = exc.details
            return {"error": details.get("writeErrors", [])[:3]}  # 只保留前 3 条错误便于定位

    def query(self, site: str, limit: int = 20):
        """只取需要的字段，避免把正文字段全量拉回来"""
        cursor = (self.col.find({"site": site}, {"title": 1, "url": 1, "pub_time": 1})
                  .sort("pub_time", -1).limit(limit))
        return list(cursor)


if __name__ == "__main__":
    store = MongoStore()
    store.ensure_index()
    print(store.upsert_many([
        {"url": "https://example.com/1", "site": "example.com",
         "title": "示例文章一", "tags": ["行业", "数据"], "pub_time": "2024-05-01T09:00:00Z"},
    ]))
    print(store.query("example.com", limit=5))

几个必须知道的细节：UpdateOne(..., upsert=True) 在「不存在则插入」和「存在则更新」两种情况下的返回值不同（前者计入 upserted_count，后者计入 modified_count），统计写入量时要把两者相加；ordered=False 允许并发写，速度更快，但错误是部分成功的，要读 BulkWriteError.details 逐条分析。

索引与查询

from pymongo import DESCENDING, TEXT

# 唯一索引：去重的最后一道防线（重复插入时会抛 DuplicateKeyError）
col.create_index([("url_md5", 1)], unique=True)

# 复合索引：注意字段顺序，等值条件在前、范围条件在后
col.create_index([("site", 1), ("collect_date", -1), ("price", 1)])

# TTL 索引：日志/原始快照自动过期，省去归档脚本
col.create_index([("created_at", 1)], expireAfterSeconds=7 * 86400)

# 文本索引：标题与正文的关键词搜索（中文分词效果有限，仅作兜底）
col.create_index([("title", TEXT), ("content", TEXT)])

# 检查索引是否被使用：executionStats 里的 totalDocsExamined 应远小于文档总数
col.find({"site": "example.com", "collect_date": "2024-05-01"}).explain("executionStats")

索引类型	用途	注意
唯一索引	保证幂等写入、去重	建索引前先清理历史重复数据，否则会失败
复合索引	等值 + 排序 + 范围查询	顺序错一个字段就退化为全表扫描
TTL 索引	原始数据自动过期	不能建在会被更新的字段上
稀疏/部分索引	只索引存在该字段的文档	适合可选字段，能显著减小索引体积

数据量变大后的两条路

归档优先：热表只留最近 90 天，历史数据 $out 或游标迁移到 news_archive_2024 集合。八成查询只关心最近数据，归档能立刻降低索引体积与扫描行数。
分片其次：sh.shardCollection("spider.news", {"url_md5": "hashed"}) 按哈希分片，写入均匀；分片键一旦选定几乎无法更改，且需要 mongos + config server 运维能力，中小规模不建议过早引入。

与 MySQL 的选型对照

维度	MongoDB	MySQL
写入路径	单文档写入，`bulk_write` 批量插入	`INSERT`/`executemany`
去重	唯一索引 + `upsert`	唯一键 + `ON DUPLICATE KEY UPDATE`
结构变更	直接写新字段	`ALTER TABLE`，大表变更代价高
事务	4.0+ 支持多文档事务，但性能有损	原生强事务
报表统计	聚合管道能力够用但不如 SQL 顺手	`GROUP BY` + 多表 JOIN 更成熟
存储成本	文档冗余，占用偏大	规范化存储更省

常见坑

现象	原因	处理
重复文档	唯一索引没建或建在 `url` 而非 `url_md5` 上	先建唯一索引再跑写入
`bulk_write` 报 DuplicateKeyError	同一批里出现重复键且 `ordered=False`	批内先去重，或改用 `ReplaceOne` 覆盖
输入中文变乱码	连接串缺 `charset` 或驱动版本过旧	使用 pymongo 4.x，默认即 UTF-8
查询突然变慢	复合索引字段顺序与查询不匹配	用 `explain("executionStats")` 核对 `totalDocsExamined`
集合越来越大	每次采集都插入新文档	明确唯一键用 `upsert` 而非 `insert_many`
时间字段无法排序	存成了字符串	存 `datetime` 对象或统一 ISO 8601 格式

合规提示：MongoDB 的灵活结构容易让人「什么都存」，反而放大了合规风险。只采集公开数据，遵守 robots.txt 与站点服务条款，控制频率与并发；不采集手机号、身份证号、简历个人信息等隐私字段，不绕过登录、验证码与付费墙。存储与使用需符合《个人信息保护法》《数据安全法》《著作权法》，敏感字段应做脱敏或不落库，优先使用官方 API 与开放数据。

小结：字段不稳定、结构嵌套时选 MongoDB，用唯一索引加 bulk_write + UpdateOne(upsert=True) 实现幂等写入，用复合索引支撑等值加排序查询；数据量增长优先归档、其次分片；需要多表 JOIN 与强事务的统计场景还是回到 MySQL。

笔记加载中…

标签：#Python3 #爬虫

返回知识库