基于DataFrame优化MongoDB批量插入前的重复数据校验性能

原方案把100万条数据存进列表逐条校验，本质是O(n*m)的时间复杂度，效率极低。用Pandas的向量化操作可将校验复杂度降至O(n)，能大幅压缩耗时。

具体实现步骤 #

1. 仅读取MongoDB中已有的queryString主键：无需拉取全量数据，只提取需要的主键字段，节省内存与IO时间。
2. 将新数据转为DataFrame：便于后续执行向量化筛选操作。
3. 快速筛选不存在的记录：用isin()方法批量判断新数据的主键是否存在于现有数据中。
4. 批量插入筛选后的新数据：保留原有的高效插入逻辑。

完整代码示例 #

import pandas as pd
from pymongo import MongoClient

# 1. 连接MongoDB
client = MongoClient("mongodb://localhost:27017/")
db = client["your_database_name"]
collection = db["your_collection_name"]

# 2. 读取现有数据的queryString主键，存入DataFrame
# 仅返回queryString字段，减少数据传输量
existing_query_strings = collection.find(
    {"source_name": "your_target_source_name"},  # 原逻辑中的source_name过滤条件
    {"queryString": 1, "_id": 0}  # 只取queryString，忽略_id
)
existing_df = pd.DataFrame(list(existing_query_strings))
existing_keys = existing_df["queryString"]

# 3. 假设新数据是包含queryString的字典列表
new_data = [
    {"queryString": "key1", "field1": "value1", "source_name": "your_target_source_name"},
    {"queryString": "key2", "field2": "value2", "source_name": "your_target_source_name"},
    # ... 更多新数据
]

# 4. 新数据转为DataFrame
new_df = pd.DataFrame(new_data)

# 5. 筛选出queryString不在现有数据中的记录
# ~表示取反，保留不存在的记录
to_insert_df = new_df[~new_df["queryString"].isin(existing_keys)]

# 6. 批量插入到MongoDB
if not to_insert_df.empty:
    collection.insert_many(to_insert_df.to_dict("records"))
    print(f"成功插入{len(to_insert_df)}条新数据")
else:
    print("没有新数据需要插入")

关键优化点 #

• 减少数据读取量：100万条主键数据占用内存极小，远小于全量数据，大幅降低IO耗时。
• 向量化操作：isin()是Pandas底层用C实现的向量化判断，比Python循环快数个数量级，3000条新数据的校验几乎瞬间完成。
• 批量插入：沿用insert_many高效插入逻辑，不影响插入阶段的速度。

额外优化建议 #

如果queryString字段未在MongoDB中创建索引，建议先创建唯一索引：

# 创建queryString的唯一索引，提升后续查询性能
collection.create_index("queryString", unique=True)

后续即使直接用MongoDB的批量插入处理重复键错误，也能避免重复数据，但提前用DataFrame筛选可减少数据库的错误请求，进一步提升效率。

内容的提问来源于stack exchange，提问作者Abcd Efgh