PyMongo 游标batch_size

Question

使用 PyMongo 3.7.2，我尝试通过在 MongoDB 游标上使用 batch_size 来分块读取集合，如here所述。基本思想是在集合对象上使用find()方法，以batch_size为参数。但是无论我尝试什么，游标总是返回我集合中的所有文档。

我的代码的一个基本片段如下所示（该集合有超过 10K 的文档）：

import pymongo as pm

client = pm.MongoClient()

coll = client.get_database('db').get_collection('coll')

cur = coll.find({}, batch_size=500)

但是，游标总是立即返回完整的集合大小。我正在按照文档中的描述使用它。

有谁知道我如何正确地批量迭代集合？有多种方法可以遍历 find() 方法的输出，但这仍会首先获取完整集合，并且只会遍历内存中已拉取的文档。batch_size 参数应该每次都获取一个批次并往返于服务器，以节省内存空间。

Answer 1

Pymongo 为Cursor类提供了一些生活质量助手，因此它会自动为您进行批处理，并将结果以文档形式返回给您。该batch_size设定完成，但这个想法是，你只需要设置它的find()方法，而不必做手工低水平电话或通过批次迭代。例如，如果我的集合中有 100 个文档：> db.test.count()100然后我设置分析级别以记录所有查询：> db.setProfilingLevel(0,-1){&nbsp; "was": 0,&nbsp; "slowms": 100,&nbsp; "sampleRate": 1,&nbsp; "ok": 1,...然后我使用 pymongo 指定batch_size10：import pymongoimport bsonconn = pymongo.MongoClient()cur = conn.test.test.find({}, {'txt':0}, batch_size=10)print(list(cur))运行该查询，我在 MongoDB 日志中看到：2019-02-22T15:03:54.522+1100 I COMMAND&nbsp; [conn702] command test.test command: find { find: "test", filter: {} ....2019-02-22T15:03:54.523+1100 I COMMAND&nbsp; [conn702] command test.test command: getMore { getMore: 266777378048, collection: "test", batchSize: 10, ....&nbsp;(getMore repeated 9 more times)所以查询是以指定的批次从服务器获取的。它只是通过Cursor课程对您隐藏。编辑如果真的需要批量获取文档，find_raw_batches()Collection下面有一个功能（doc link）。此方法的工作方式find()与此类似，并接受相同的参数。但是请注意，它将返回需要由应用程序在单独的步骤中解码的原始 BSON。值得注意的是，此方法不支持session。话虽如此，如果目标是降低应用程序的内存使用量，则值得考虑修改查询，使其使用范围。例如：find({'$gte': <some criteria>, '$lte': <some other criteria>})范围查询更容易优化，可以使用索引，并且（在我看来）更容易调试和更容易在查询中断时重新启动。这在使用批处理时不太灵活，您必须从头开始重新启动查询，如果它被中断，则再次检查所有批处理。

Answer 2

我就是这样做的，它有助于将数据分块，但我认为会有更直接的方法来做到这一点。我创建了一个 yield_rows 函数，它可以让您生成和生成块，它确保删除使用的块。import pymongo as pmCHUNK_SIZE = 500client = pm.MongoClient()coll = client.get_database('db').get_collection('coll')cursor = coll.find({}, batch_size=CHUNK_SIZE)def yield_rows(cursor, chunk_size):&nbsp; &nbsp; """&nbsp; &nbsp; Generator to yield chunks from cursor&nbsp; &nbsp; :param cursor:&nbsp; &nbsp; :param chunk_size:&nbsp; &nbsp; :return:&nbsp; &nbsp; """&nbsp; &nbsp; chunk = []&nbsp; &nbsp; for i, row in enumerate(cursor):&nbsp; &nbsp; &nbsp; &nbsp; if i % chunk_size == 0 and i > 0:&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; yield chunk&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; del chunk[:]&nbsp; &nbsp; &nbsp; &nbsp; chunk.append(row)&nbsp; &nbsp; yield chunkchunks = yield_rows(cursor, CHUNK_SIZE)for chunk in chunks:&nbsp; &nbsp; # do processing here&nbsp; &nbsp; pass如果我找到一种更清洁、更有效的方法来做到这一点，我会更新我的答案。