python - Mongodb批量写入Updateone或Updatemany

Question

我想知道使用 updateone 或 updatemany 进行批量写入是否更快(导入)。我使用 pymongo 将数据导入集合的代码如下:

for file in sorted_files:
    df = process_file(file)
    for row, item in df.iterrows():
        data_dict = item.to_dict()
        bulk_request.append(UpdateOne(
            {"nsamples": {"$lt": 12}},
            {
                "$push": {"samples": data_dict},
                "$inc": {"nsamples": 1}
            },
            upsert=True
        ))
    result = mycol1.bulk_write(bulk_request)

当我尝试更新许多内容时，我唯一改变的是:

...
...
bulk_request.append(UpdateMany(..
..
..

我没有发现插入时间有任何重大差异。updateMany 不应该更快吗？ 也许我做错了什么。任何建议都会有帮助! 提前致谢!

注意:我的数据由 120 万行组成。我需要每个文档包含 12 个子文档。

Answer 1

@Wernfried Domscheit's答案是正确的。

此答案特定于您的场景。

如果您不介意不将记录更新到现有文档并完全插入新文档，请使用以下最适合您的用例的代码。

sorted_files = []
process_file = None
for file in sorted_files:
    df = process_file(file)
    sample_data = []
    for row, item in df.iterrows():
        sample_data.append(item.to_dict())
        if len(sample_data) == 12:
            mycol1.insertOne({
                "samples": sample_data,
                "nsamples": len(sample_data),
            })
            sample_data = []
    mycol1.insertOne({
        "samples": sample_data,
        "nsamples": len(sample_data),
    })

如果您想用 12 个对象填充现有记录，然后， 创建新记录，使用以下代码逻辑。

Note: I have not tested the code in my local, its just to understand the flow for you to use.

for file in sorted_files:
    df = process_file(file)
    sample_data = []
    continuity_flag = False
    for row, item in df.iterrows():
        sample_data.append(item.to_dict())
        if not continuity_flag:
            sample_rec = mycol1.find_one({"nsamples": {"$lt": 12}}, {"nsamples": 1})
            if sample_rec is None:
                continuity_flag = True
            elif sample_rec["nsamples"] + len(sample_data) == 12:
                mycol1.update_one({
                    "_id": sample_rec["_id"]
                }, {
                    "$push": {"samples": {"$each": sample_data}},
                    "$inc": {"nsamples": len(sample_data)}
                })
        if len(sample_data) == 12:
            mycol1.insert_one({
                "samples": sample_data,
                "nsamples": len(sample_data),
            })
            sample_data = []
    if sample_data:
        mycol1.insert_one({
            "samples": sample_data,
            "nsamples": len(sample_data),
        })