PyMongo 带有生成器的批量写入操作功能

Question

我想使用 PyMongo 的 bulk 执行写操作的写操作特性 以减少网络往返次数并增加 rite 吞吐量。

我还发现here可以用5000作为批号。

但是，我不想要批号的最佳大小以及如何在下面的代码中将 PyMongo 的批量写入操作功能与生成器结合起来？

from pymongo import MongoClient
from itertools import groupby
import csv


def iter_something(rows):
    key_names = ['type', 'name', 'sub_name', 'pos', 's_type', 'x_type']
    chr_key_names = ['letter', 'no']
    for keys, group in groupby(rows, lambda row: row[:6]):
        result = dict(zip(key_names, keys))
        result['chr'] = [dict(zip(chr_key_names, row[6:])) for row in group]
        yield result


def main():
    converters = [str, str, str, int, int, int, str, int]
    with open("/home/mic/tmp/test.txt") as c:
    reader = csv.reader(c, skipinitialspace=True)
    converted = ([conv(col) for conv, col in zip(converters, row)] for row in reader)
    for object_ in iter_something(converted):
        print(object_)


if __name__ == '__main__':
    db = MongoClient().test
    sDB = db.snps 
    main()

test.txt 文件：

  Test, A, B01, 828288,  1,    7, C, 5
  Test, A, B01, 828288,  1,    7, T, 6
  Test, A, B01, 171878,  3,    7, C, 5
  Test, A, B01, 171878,  3,    7, T, 6
  Test, A, B01, 871963,  3,    9, A, 5
  Test, A, B01, 871963,  3,    9, G, 6
  Test, A, B01, 1932523, 1,   10, T, 4
  Test, A, B01, 1932523, 1,   10, A, 5
  Test, A, B01, 1932523, 1,   10, X, 6
  Test, A, B01, 667214,  1,   14, T, 4
  Test, A, B01, 667214,  1,   14, G, 5
  Test, A, B01, 67214,   1,   14, G, 6      

Answer 1

因此，您有一个文档生成器，并且您希望将其拆分为文档块或文档组。这可以使用grouper 生成器优雅地完成，在this answer 中进行了描述。

然后，对于每组文档，使用 pymongo 的 insert 批量插入它们。

你得到：

def main():
    db = MongoClient().test
    sDB = db.snps 
    ...
    for docs_group in grouper(iter_something(converted), BULK_SIZE):
        docs_group = [ doc for doc in docs_group if doc is not None ]  # filter out Nones
        sDB.insert(docs_group, ...)

关于最佳 BULK_SIZE，这取决于各种因素，例如典型的文档大小、网络延迟等。您需要进行试验。

Answer 2

你可以这样做：

sDB.insert(iter_something(converted))

PyMongo 会做正确的事：迭代您的生成器，直到它产生 1000 个文档或 16MB 数据，然后在将批处理插入 MongoDB 时暂停生成器。插入批次后，PyMongo 将恢复您的生成器以创建下一个批次，并继续直到插入所有文档。然后 insert() 返回插入的文档 ID 列表。

在 this commit PyMongo 中添加了对生成器的初始支持，此后我们一直支持文档生成器。