pymongo 光标“触摸”以避免超时

Posted 2023-03-08

【中文标题】pymongo 光标“触摸”以避免超时

【英文标题】：pymongo cursor 'touch' to avoid timeout

【发布时间】：2017-03-18 19:48:21

【问题描述】：

我需要从 mongo (v3.2.10) 集合（使用 Pymongo 3.3.0）中获取大量（例如 1 亿个）文档并对其进行迭代。迭代需要几天时间，而且我经常因为光标超时而遇到异常。

在我的情况下，我需要在迭代时睡眠不可预测的时间。因此，例如，我可能需要： - 获取 10 个文件 - 睡眠 1 秒 - 获取 1000 个文档 - 睡 4 小时 - 获取 1 个文档 等等

我知道我可以：

完全禁用超时，但我希望尽可能避免这种情况，因为如果我的代码完全停止运行，我很高兴为我清理游标 减小我的光标的 batch_size，但这不会有帮助，例如如上例所示，我需要睡 4 小时

似乎一个不错的解决方案是“触摸”光标以使其保持活动状态。因此，例如，我会将长时间的睡眠分成更短的间隔，并在每个间隔之间触摸光标。

我没有看到通过 pymongo 执行此操作的方法，但我想知道是否有人确切知道这是否可能。

【参考方案1】：

当然，这是不可能的，你想要的是功能SERVER-6036，这是未实现的。

对于这样一个长时间运行的任务，我建议对索引字段进行查询。例如。如果您的文档都有时间戳“ts”：

documents = list(collection.find().sort('ts').limit(1000))
for doc in documents:
    # ... process doc ...

while True:
    ids = set(doc['_id'] for doc in documents)
    cursor = collection.find('ts': '$gte': documents[-1]['ts'])
    documents = list(cursor.limit(1000).sort('ts'))
    if not documents:
        break  # All done.
    for doc in documents:
        # Avoid overlaps
        if doc['_id'] not in ids:
            # ... process doc ...

此代码完全迭代光标，因此不会超时，然后处理 1000 个文档，然后重复下一个 1000 个。

第二个想法：用a very long cursor timeout配置你的服务器：

mongod --setParameter cursorTimeoutMillis=21600000  # 6 hrs

第三个想法：虽然不完全确定，您可以更确定您将通过在with 语句中使用非超时游标来关闭它：

cursor = collection.find(..., no_cursor_timeout=True)
with cursor:
    # PyMongo will try to kill cursor on server
    # if you leave this block.
    for doc in cursor:
        # do stuff....

【讨论】：

是的，这种模式在可能的情况下很有用，但在我的情况下，我有一些非常复杂的查询，我不想每次都用新的查询重新开始 - 我想保留一个光标以某种方式在服务器上打开

我已经更新了我的答案。您问“是否有人确切知道这是否可能”，我已经明确回答。 =)

您的回答确实很完整，所以我接受了。我想我只是有一个堆栈溢出的烦恼，因为每次我问一个重点问题“我该怎么做 X”时，我总是会收到一百万个愤怒的回答说“你不想做 X 你应该做 Y 或 Z”当然，在我的情况下 Y 或 Z 是不可能的，但要解释这一点需要太多的上下文:)。不管怎样，谢谢你的回答，它确实很有帮助！

我理解您的烦恼，感谢您接受我的回答。 =)

【参考方案2】：

对我来说，甚至 no_cursor_timeout=True 都没有工作，所以我创建了一个函数，将光标中的数据保存在一个临时文件中，然后将文件从文件中返回给调用者。

from tempfile import NamedTemporaryFile
import pickle
import os

def safely_read_from_cursor(cursor):
    # save data in a local file
    with NamedTemporaryFile(suffix='.pickle', prefix='data_', delete=False) as data_file, cursor:
        for count, doc in enumerate(cursor, 1):
            pickle.dump(doc, data_file)

    # open file again and iterate over data
    with open(data_file.name, mode="rb") as data_file:
        for _ in range(count):
            yield pickle.load(data_file)

    # remove temporary file
    os.remove(data_file.name)