使用 PyMongo 在 MongoDB 中进行批量插入/更新答案

【问题标题】：Do bulk inserts/update in MongoDB with PyMongo使用 PyMongo 在 MongoDB 中进行批量插入/更新
【发布时间】：2017-07-12 22:07:48
【问题描述】：

如何使用 pymongo/pandas 在 mongoDb 中批量更新/插入。我得到的错误是batch op errors occurred 我之所以得到是因为我设置了我想做的"_id"。我的代码在第一次运行时运行良好，但在第二次运行时失败。我想在工作流程中使用熊猫。数据确实有一个日期时间对象。

upsert = True 的语法与Update 完全不同。使用update 的有效解决方案会有所帮助，其中可以设置"_id" 或"qid"。 但是，有 python 日期时间对象！

InSQL   = 'SELECT * from  database2.table2 '
sqlOut  = pd.read_sql(InSQL,cxn)
sqlOut['_id'] = "20170101" + ":"+ sqlOut['Var']   

dfOut   = sqlOut.to_json(orient='records',date_format='iso' )
try:
    db["test"].insert_many(json.loads(dfOut))
except Exception as e:  print e

我提供了 50pt 赏金，但已过期，但没有任何答复。嗯……

【问题讨论】：

Mongo 使用了string base input 并且不允许任何变量插入（仅限运算符）check this, json.loads(dfOut) 你不能插入局部变量作为入口！主键YYYYmmDDHHMMSS+Counter value
您可以深入了解BulkWriteError 的详细信息并更好地了解正在发生的事情。可能是您个人定义的 id 重复或违反了 12 字节限制。详情请见stackoverflow.com/questions/30355790/mongodb-bulk-write-error。
_id 第一次工作的事实告诉我，12 字节的限制不是问题。是的，_id 或 gid 是重复的，需要更新。
听起来您的目标是执行 upsert，但当前示例代码正在执行插入（因此预计连续运行将因重复的 _id 键而失败）。你能用dfOut 内容和你尝试的upsert 代码的简短示例进行更新吗？另外，您使用的是什么版本的 PyMongo？我怀疑您实际上想使用 update_many() 和 upsert 选项集。您能否澄清一下对 Python 日期时间对象和 _id 与 qid 的担忧？是否要将qid 用作_id？

标签： python sql mongodb pandas pymongo

【解决方案1】：

您收到错误消息，因为您尝试在第二次和随后的insert_many 调用中插入字段与现有文档的字段冲突的文档。您正确推断出这可能是由于您明确设置了_id，这将与集合中现有的_id 值冲突。

MongoDB automatically creates an unique index on _id，禁止重复值。

您需要更新或替换您的文档在第一个（将文档插入到他们的第一个版本）之后的调用。确实有一个“upsert”的概念，它将负责在集合中插入不存在的文档以及更新现有的文档。

您的选择：

最高效：pymongo.collection.Collection.bulk_write

import pymongo

operations = [pymongo.operations.ReplaceOne(
    filter={"_id": doc["_id"]}, 
    replacement=doc,
    upsert=True
    ) for doc in json.loads(dfOut)]

result = db["test"].bulk_write(operations)
# handle results

请注意，它的效率还取决于该字段是否在集合中被索引，顺便提一下_id 的情况。（另见pymongo.operations.ReplaceOne）

遍历您的集合并调用pymongo.collection.Collection.update_one 或pymongo.collection.Collection.replace_one（效率低，因为不是批量）

import pymongo

results = []
for doc in json.load(dfOut):
    result = db["test"].replace_one(
        filter={"_id": doc["_id"]},
        replacement=doc,
        upsert=True
        )
    results.append(result)

# handle results

注意：pymongo.collection.Collection.update_many 似乎不适合您的需求，因为您没有尝试在给定过滤器的所有匹配项上设置相同的值。

【讨论】：

【解决方案2】：

批量操作错误可能是由重复的_id引起的，所以在插入之前删除 mongo 中已经存在的相同 _id 文档

或者使用 update_many https://api.mongodb.com/python/current/api/pymongo/collection.html?highlight=update#pymongo.collection.Collection.update_many

https://docs.mongodb.com/manual/reference/method/db.collection.updateMany/

【讨论】：