4

我正在更新一个包含数百万个文档且 _id 冲突少于 10 个的数据库。

我目前正在使用 PyMongo 模块通过 insert_many 进行批量插入：

  1. 查询db看_id是否存在
  2. 如果 _id 不存在，则将文档添加到数组中
  3. 使用 insert_many 插入数据库，一次插入 1000 个文档。

几百万个文档中只有大约 10 个冲突，我目前正在为每个 _id 查询数据库。我认为如果我可以减少查询过程，我可以将整体插入时间减少一两天。

是否有类似于 upsert 的东西，如果它不存在，它可能只插入一个文档？

4

2 回答 2

9

处理此问题并以有效方式“插入/更新”许多文档的更好方法是使用批量操作 API “批量”提交所有内容，并有效发送所有内容并在确认中接收“单一响应”。

这可以通过两种方式处理。

首先要忽略主键或其他索引上的任何“重复错误”，然后您可以使用“UnOrdered”形式的操作：

bulk = pymongo.bulk.BulkOperationBuilder(collection,ordered=False)
for doc in docs:
    bulk.insert(doc)

response = bulk.execute()

那里的“UnOrdered”或false参数意味着操作都可以按任何顺序执行，并且“整个”批次将完成，任何实际错误都将在响应中“报告”。所以这是一种基本上“忽略”重复并继续前进的方法。

替代方法大致相同，但使用“upsert”功能以及$setOnInsert：

bulk = pymongo.bulk.BulkOperationBuilder(collection,ordered=True)
for doc in docs:
    bulk.find({ "_id": doc["_id"] }).upsert().updateOne({
        "$setOnInsert": doc
    })

response = bulk.execute()

其中“查询”部分.find()用于使用“主键”或文档的“唯一键”查询文档的存在。如果未找到匹配项，则会在创建新文档时发生“更新插入”。由于所有修改内容都在$setOnInsert其中，因此只有在发生“更新插入”时才会在此处修改文档字段。否则，当文档“匹配”时，对于保存在此​​运算符下的数据，实际上没有任何更改。

在这种情况下，“有序”意味着每个语句实际上都是按照创建它的“相同”顺序提交的。此外，这里的任何“错误”都会停止更新（在发生错误的地方），这样就不会再有任何操作了做出承诺。它是可选的，但可能建议用于正常的“复制”行为，其中后面的语句“复制”前一个的数据。

因此，为了更高效的写入，总体思路是使用“批量”API 并相应地构建您的操作。这里的选择实际上归结为来自源的“插入顺序”对您是否重要。

当然，相同的"ordered"=False操作适用insert_many于在较新的驱动程序版本中实际使用“批量”操作的操作。但是你会因为坚持使用可以“混合”操作和简单 API 的通用接口而获得更多的灵活性。

于 2015-07-13T05:14:28.530 回答
2

虽然 Blakes 的回答很好，但在大多数情况下，使用ordered=False参数并捕获BulkWriteError重复项是可以的。

try:
    collection.insert_many(data, ordered=False)
except BulkWriteError:
    logger.info('Duplicates were found.')
于 2019-05-27T03:47:38.410 回答