MongoDB插入文档超16MB报错的非GridFS解决方案咨询

好的，针对你遇到的MongoDB单文档超过16MB限制的问题，又不想用GridFS的话，我们可以从缩小文档体积和拆分文档结构两个方向来解决，下面具体说下可行的方案：

方案一：压缩或精简大字段内容 #

你的文档里tokens、tokens_missing和token_mapping是体积大头，如果你不需要对这些字段里的内容做MongoDB原生查询（比如用$in匹配tokens里的元素），可以通过压缩来减小体积：

示例：用zlib压缩大列表/字典 #

import zlib
import pickle
from pymongo import MongoClient

conn = MongoClient("mongodb://sample_mongo:27017")
db_conn = conn["test"]
db_collection = db_conn["sample"]

# 原始大记录
record = {
    "name": "drugs",
    "collection_id": 23,
    "timestamp": 1515065002,
    "tokens": [], # 大列表
    "tokens_missing": [], # 大列表
    "token_mapping": {} # 大字典
}

# 压缩大字段
record["tokens_compressed"] = zlib.compress(pickle.dumps(record["tokens"]))
record["tokens_missing_compressed"] = zlib.compress(pickle.dumps(record["tokens_missing"]))
record["token_mapping_compressed"] = zlib.compress(pickle.dumps(record["token_mapping"]))

# 删除原始大字段
del record["tokens"], record["tokens_missing"], record["token_mapping"]

# 插入文档
db_collection.insert(record, check_keys=False)

# 读取时解压
retrieved_doc = db_collection.find_one({"collection_id": 23})
original_tokens = pickle.loads(zlib.decompress(retrieved_doc["tokens_compressed"]))
original_mapping = pickle.loads(zlib.decompress(retrieved_doc["token_mapping_compressed"]))

额外优化点 #

• 如果tokens里有大量重复字符串，可以先做去重或者用ID映射（比如把重复token存到另一个字典，用ID代替原字符串，减少冗余）
• 对于token_mapping，如果键值对有规律，也可以用更紧凑的存储格式（比如把字典转成列表存，读取时再转回去）

方案二：拆分文档到关联集合（推荐） #

如果需要保留对大字段内容的查询能力，最可靠的方式是把大字段拆分到单独的集合，通过关联字段（比如你的collection_id）来关联查询，这也是MongoDB推荐的非GridFS大文档处理方式：

步骤1：拆分插入文档 #

from pymongo import MongoClient

conn = MongoClient("mongodb://sample_mongo:27017")
db_conn = conn["test"]
main_collection = db_conn["sample"]
token_collection = db_conn["sample_tokens"] # 新建存储大字段的集合

# 主文档只保留小体积字段
main_record = {
    "name": "drugs",
    "collection_id": 23,
    "timestamp": 1515065002
}
main_collection.insert(main_record, check_keys=False)

# 大字段单独存入关联集合
token_record = {
    "collection_id": 23, # 关联键，和主文档对应
    "tokens": [], # 原始大列表
    "tokens_missing": [], # 原始大列表
    "token_mapping": {} # 原始大字典
}
token_collection.insert(token_record, check_keys=False)

步骤2：关联查询合并数据 #

需要读取完整数据时，用MongoDB的$lookup聚合操作关联两个集合：

pipeline = [
    {"$match": {"collection_id": 23}}, # 匹配主文档
    {"$lookup": {
        "from": "sample_tokens",
        "localField": "collection_id",
        "foreignField": "collection_id",
        "as": "token_data"
    }}
]

# 执行聚合查询并合并数据
result = list(main_collection.aggregate(pipeline))
if result:
    merged_doc = result[0]
    # 取出关联的token数据（因为$lookup返回数组，取第一个元素）
    token_data = merged_doc.pop("token_data")[0]
    merged_doc.update(token_data)
    print(merged_doc) # 这就是合并后的完整文档

方案三：拆分嵌套子文档（限大字段可分割场景） #

如果你的大字段可以分割成多个小部分，也可以把它们拆成主文档内的嵌套子文档数组，但注意总文档体积还是不能超过16MB，所以只适合大字段拆分后总容量可控的情况：

from pymongo import MongoClient

conn = MongoClient("mongodb://sample_mongo:27017")
db_conn = conn["test"]
db_collection = db_conn["sample"]

record = {
    "name": "drugs",
    "collection_id": 23,
    "timestamp": 1515065002,
    "tokens": [], # 原始大列表
    "tokens_missing": [], # 原始大列表
    "token_mapping": {} # 原始大字典
}

# 把tokens拆分成每1000个元素一组的子文档
split_tokens = [{"batch": i, "data": record["tokens"][i:i+1000]} for i in range(0, len(record["tokens"]), 1000)]
record["tokens_batches"] = split_tokens
del record["tokens"]

# 同理拆分tokens_missing
split_missing = [{"batch": i, "data": record["tokens_missing"][i:i+1000]} for i in range(0, len(record["tokens_missing"]), 1000)]
record["tokens_missing_batches"] = split_missing
del record["tokens_missing"]

# 如果token_mapping太大，可以按键的前缀或者其他规则拆分成多个小字典
# 比如按首字母拆分token_mapping
mapping_batches = {}
for key, value in record["token_mapping"].items():
    first_char = key[0].lower()
    if first_char not in mapping_batches:
        mapping_batches[first_char] = {}
    mapping_batches[first_char][key] = value
record["token_mapping_batches"] = mapping_batches
del record["token_mapping"]

# 插入文档
db_collection.insert(record, check_keys=False)

总结一下：如果需要查询大字段内容，优先用方案二（关联集合）；如果不需要查询大字段内部内容，用**方案一（压缩）**更简单；方案三只适合大字段能被均匀分割的场景。

内容的提问来源于stack exchange，提问作者Thrisundar Reddy J