在使用 Python 操作 OpenSearch 时，opensearchpy 是官方提供的低层客户端库。很多人在写完一个 match 查询后发现，无论索引里有多少符合条件的文档，客户端返回的结果列表总是只有一部分。这是因为 OpenSearch 的搜索接口本身就有分页机制，默认一次只返回十条记录。如果你的业务逻辑要求把某个查询条件下的所有文档都取出来做离线分析、批量迁移或者全量导出，就必须使用能够跨越分页边界连续拉取数据的方案。opensearchpy 提供了多种手段来实现这个目标，其中最经典的是 scroll 滚动查询，而在新版本中更推荐用 point in time 配合 search_after。下面我们先看一个基础但完整的 scroll 用法示例。
使用 scroll 滚动查询获取全量结果
scroll 的核心思想是：第一次搜索时，OpenSearch 会在服务端创建一个快照上下文，并保留一段时间。客户端拿到第一批数据和一串 scroll_id，之后不断拿着这个 scroll_id 去请求下一页，直到没有数据返回为止。这种方式适合数据量较大、且对实时性要求不高的后台任务。在 opensearchpy 中，可以通过 helpers.scan 函数极其简洁地实现，也可以手动调用 search 与 scroll 方法精细控制。
下面是一个手动控制 scroll 的示例，我们设置 scroll 有效期为五分钟，每次拉取一百条。注意在循环里要及时用返回的新 scroll_id 替换旧的，并在结束后调用 clear_scroll 释放服务端资源，否则会造成上下文堆积。
from opensearchpy import OpenSearch
client = OpenSearch(
hosts=[{'host': 'localhost', 'port': 9200}],
http_auth=('admin', 'admin')
)
query = {
'query': {
'match': {
'status': 'active'
}
}
}
resp = client.search(
index='orders',
body=query,
scroll='5m',
size=100
)
scroll_id = resp['_scroll_id']
hits = resp['hits']['hits']
while len(hits) > 0:
for doc in hits:
print(doc['_id'], doc['_source'])
resp = client.scroll(scroll_id=scroll_id, scroll='5m')
scroll_id = resp['_scroll_id']
hits = resp['hits']['hits']
client.clear_scroll(scroll_id=scroll_id)
上面的代码虽然能跑通，但在生产环境里更推荐使用 opensearchpy.helpers.scan，它会自动处理 scroll_id 的续期和清空动作，代码量更少也更安全。scan 函数返回一个生成器，你可以用 for 循环直接遍历所有命中文档，底层自动分批拉取。使用 scan 时，要通过 query 参数传入检索体，用 size 控制每批大小，用 scroll 控制上下文存活时间。
不过 scroll 也有明显短板。由于它在查询开始时锁定了索引状态的快照，如果索引在滚动期间持续写入，新文档不会被扫到；同时快照会占用堆内存，并发多个大 scroll 容易把节点压垮。因此对于需要一致性和低开销的场景，应该考虑下面的 PIT 方案。
基于 point in time 与 search_after 的取数方式
point in time（简称 PIT）是 OpenSearch 后来引入的能力，它在不锁定整个索引快照的前提下，提供一个轻量的时间点视图。配合 search_after 参数，可以实现深分页式的全量遍历，且资源占用远低于 scroll。使用流程是：先创建 PIT 拿到 pit_id，再在 search 请求里带上 pit 和 sort 规则，用上一次结果里最后一个文档的排序值作为 search_after 继续查，直到某次返回为空。
下面的例子展示了如何用 opensearchpy 创建 PIT 并循环拉取。注意排序字段必须是唯一或可重复的复合键，否则 search_after 会漏数据或死循环。通常我们用 _shard_doc 这个内置字段来保证全局顺序。
from opensearchpy import OpenSearch
client = OpenSearch(
hosts=[{'host': 'localhost', 'port': 9200}],
http_auth=('admin', 'admin')
)
pit = client.create_point_in_time(index='orders', keep_alive='1m')
pit_id = pit['pit_id']
body = {
'size': 100,
'query': {
'match': {
'status': 'active'
}
},
'pit': {
'id': pit_id,
'keep_alive': '1m'
},
'sort': [
{'_shard_doc': 'asc'}
]
}
resp = client.search(body=body)
hits = resp['hits']['hits']
while len(hits) > 0:
for doc in hits:
print(doc['_id'], doc['_source'])
last_sort = hits[-1]['sort']
body['search_after'] = last_sort
resp = client.search(body=body)
hits = resp['hits']['hits']
client.delete_point_in_time(body={'pit_id': pit_id})
与 scroll 相比，PIT 加 search_after 不会在服务器端长期保留大块快照内存，只维护一个轻量视图，因此在超大数据集和并发导出时更加稳健。缺点是代码稍微复杂，且要求每次查询都带一致的 sort 规则。如果你的 OpenSearch 版本较老不支持 PIT，那只能退回 scroll 方案。
在实际项目中，还可以把两种方案封装成一个通用迭代器，根据集群版本自动选择。无论哪种方式，都不要试图把 size 设得极大然后一次性 search，那样极易触发网关超时和内存熔断。
借助 helpers 模块简化全量抓取逻辑
opensearchpy 的 helpers 子模块专门封装了批量操作工具，其中的 scan 函数就是为“获取查询所有结果”而生的。它默认使用 scroll，但把续期、翻页、清理都隐藏在生成器里。对于绝大多数离线脚本来说，直接调用 scan 是最省心的做法，不需要自己管理 scroll_id，也不容易写出资源泄漏的 bug。
下面的示例只用几行就能完成全量遍历。scan 的第一个参数是客户端，index 指定索引，query 放 DSL，size 控制每批条数。你在循环里拿到的每个元素就是一条 hit 字典，包含 _id 和 _source 等字段。如果索引文档非常多，scan 会在后台默默分批拉取，对调用方完全透明。
from opensearchpy import OpenSearch
from opensearchpy.helpers import scan
client = OpenSearch(
hosts=[{'host': 'localhost', 'port': 9200}],
http_auth=('admin', 'admin')
)
query = {
'query': {
'match_all': {}
}
}
for doc in scan(client, index='orders', query=query, size=500, scroll='2m'):
print(doc['_id'], doc['_source'])
虽然 scan 很方便，但它底层仍是 scroll，因此同样有快照一致性和内存占用问题。如果你需要的是“当前最新状态的全量”而不是“某个旧快照的全量”，且集群版本允许，建议把 helpers 的源码稍微改一下改用 PIT，或者自己按上一节的写法封装。另外，scan 不支持聚合结果的全量拉取，它只适用于普通搜索命中的文档流。
总结来看，获取 opensearchpy 查询的所有结果并不是简单地调一次 search，而是要根据数据规模、实时性要求和集群版本，选择 scroll 或 PIT 加 search_after。对于快速写脚本，用 helpers.scan 最省力；对于严谨的生产管道，手写 PIT 循环更可控。理解这两种机制的底层差异，才能在导出海量数据时既拿得全又不拖垮集群。
opensearchpyOpenSearchscroll修改时间：2026-08-15 09:54:35