寻求Python(PyMongo)批量更新MongoDB中priority字段的方法

选择update_one()还是update_many()？ #

肯定用**update_many()**（或批量操作bulk_write()）。update_one()仅能更新单个文档，循环调用它处理10000+用户会产生大量冗余网络请求，效率极低；而update_many()可以一次性匹配所有目标文档并完成更新，是批量场景的最优选择。

实现方案 #

根据数据存储场景，有两种主流实现方式：

方案1：直接在MongoDB中批量更新（推荐）

如果数据已存储在MongoDB中，无需先加载到Pandas，直接通过数据库操作完成更新，效率最高。

假设你的文档结构如下（以用户ID和偏好字段为例）：

{
  "user_id": "user_123",
  "preferences": {
    "P1": {"type": "normal", "priority": 2},
    "P2": {"type": "normal", "priority": 1}
  }
}

代码实现：

from pymongo import MongoClient

def update_priorities_in_db():
    # 连接MongoDB
    client = MongoClient('mongodb://localhost:27017/')
    db = client['你的数据库名']
    mycol = db['你的集合名']
    
    # 定义更新规则：priority=1→2，priority=2→3，3保持不变
    update_rule = {
        "$set": {
            "preferences": {
                # 将嵌套字典转为数组，遍历修改后再转回字典
                "$arrayToObject": {
                    "$map": {
                        "input": {"$objectToArray": "$preferences"},
                        "as": "item",
                        "in": {
                            "k": "$$item.k",
                            "v": {
                                "type": "$$item.v.type",
                                "priority": {
                                    "$switch": {
                                        "branches": [
                                            {"case": {"$eq": ["$$item.v.priority", 1]}, "then": 2},
                                            {"case": {"$eq": ["$$item.v.priority", 2]}, "then": 3}
                                        ],
                                        "default": "$$item.v.priority"
                                    }
                                }
                            }
                        }
                    }
                }
            }
        }
    }
    
    # 执行批量更新（第一个参数为过滤条件，{}表示匹配所有文档）
    result = mycol.update_many({}, update_rule)
    print(f"成功更新 {result.modified_count} 个文档")
    
    client.close()

# 调用函数
update_priorities_in_db()

如果你的P1、P2是顶级字段而非嵌套在preferences下，可直接针对每个字段单独定义更新规则，示例：

update_rule = {
    "$set": {
        "P1.priority": {
            "$switch": {
                "branches": [
                    {"case": {"$eq": ["$P1.priority", 1]}, "then": 2},
                    {"case": {"$eq": ["$P1.priority", 2]}, "then": 3}
                ],
                "default": "$P1.priority"
            }
        },
        "P2.priority": {
            "$switch": {
                "branches": [
                    {"case": {"$eq": ["$P2.priority", 1]}, "then": 2},
                    {"case": {"$eq": ["$P2.priority", 2]}, "then": 3}
                ],
                "default": "$P2.priority"
            }
        }
    }
}

方案2：先处理Pandas DataFrame再同步到数据库

如果需要先通过Pandas对数据做其他处理，再同步更新到MongoDB，可按以下步骤实现：

import pandas as pd
from pymongo import MongoClient, UpdateOne

def process_df_and_sync_db(user_df):
    # 定义单个单元格的更新函数
    def modify_priority(cell):
        for key in cell:
            if cell[key]["priority"] == 1:
                cell[key]["priority"] = 2
            elif cell[key]["priority"] == 2:
                cell[key]["priority"] = 3
        return cell
    
    # 应用更新到DataFrame的目标列（假设列名为'preferences'）
    user_df["preferences"] = user_df["preferences"].apply(modify_priority)
    
    # 连接MongoDB
    client = MongoClient('mongodb://localhost:27017/')
    db = client['你的数据库名']
    mycol = db['你的集合名']
    
    # 准备批量更新操作（用user_id作为匹配主键）
    bulk_operations = []
    for _, row in user_df.iterrows():
        bulk_operations.append(
            UpdateOne(
                {"user_id": row["user_id"]},
                {"$set": {"preferences": row["preferences"]}}
            )
        )
    
    # 执行批量更新
    if bulk_operations:
        result = mycol.bulk_write(bulk_operations)
        print(f"成功更新 {result.modified_count} 个文档")
    
    client.close()

# 假设你的DataFrame名为user_data_df
# process_df_and_sync_db(user_data_df)

内容的提问来源于stack exchange，提问作者Mohammadreza Motallebi