OpenSearch Python Client : Scroll API는 전체 데이터 검색을 구현합니다

本教程详细介绍了如何使用opensearch-py库通过OpenSearch的Scroll API来克服单次查询10,000条结果的限制，从而高效地获取所有匹配的查询结果。文章将提供具体的Python代码示例，指导用户如何初始化客户端、构建查询、发起滚动查询并迭代处理大量数据，最终实现全量数据导出或分析。

引言：OpenSearch查询的限制与Scroll API

OpenSearch（或Elasticsearch）的search API默认情况下为了性能考虑，限制了单次查询返回的最大结果数，通常为10,000条。这意味着当您的查询匹配的数据量远超此限制时，简单的search请求无法获取所有数据。为了解决这一问题，OpenSearch提供了Scroll API。

Scroll API允许您创建一个“快照”式搜索上下文，并分批次地检索大量查询结果，而无需担心深度分页带来的性能问题。它非常适合需要全量数据导出的场景，例如数据分析、报告生成或数据迁移。

理解Scroll API的工作原理

Scroll API的工作流程分为两步：

1. 初始化滚动查询： 第一次调用client.search()方法时，除了提供查询体外，还需要指定scroll参数（例如scroll='1m'表示滚动上下文保持1分钟）。OpenSearch会返回第一批结果和一个_scroll_id。
2. 迭代获取后续结果： 随后的请求不再是search，而是调用client.scroll()方法，并传入上一次获得的_scroll_id和新的scroll时间。OpenSearch会返回下一批结果，并可能更新_scroll_id。这个过程会持续进行，直到不再有匹配的结果返回。

使用opensearch-py实现全量数据检索

以下是使用opensearch-py库通过Scroll API获取OpenSearch全量查询结果的详细步骤和示例代码。

1. 环境准备与客户端初始化

首先，确保您已安装opensearch-py库。如果未安装，请使用pip进行安装：

pip install opensearch-py

接下来，初始化OpenSearch客户端。根据您的OpenSearch集群配置，可能需要使用不同的认证方式（例如AWS签名认证或基本认证）。

import csv
from opensearchpy import OpenSearch, RequestsHttpConnection
# 如果使用AWS认证，需要安装requests-aws4auth并导入
from requests_aws4auth import AWS4Auth

# OpenSearch集群配置 (请替换为您的实际值)
host = 'your-opensearch-host.amazonaws.com'  # 例如: 'your-domain.us-east-1.es.amazonaws.com'
port = 443 # 默认端口
use_ssl = True
verify_certs = True
timeout = 300 # 请求超时时间
pool_maxsize = 20 # 连接池大小

# 认证方式示例
# 1. AWS签名认证 (适用于AWS OpenSearch Service)
region = 'your-aws-region' # 例如: 'us-east-1'
service = 'es'
# 请替换为您的AWS凭证获取方式，例如从环境变量、IAM角色或配置文件
# from botocore.session import get_session
# session = get_session()
# credentials = session.get_credentials()
# auth = AWS4Auth(credentials.access_key, credentials.secret_key, region, service, session_token=credentials.token)

# 2. 基本认证 (适用于用户名/密码认证)
auth = ('your_username', 'your_password')

# 初始化OpenSearch客户端
client = OpenSearch(
    hosts=[{"host": host, "port": port}],
    http_auth=auth,
    use_ssl=use_ssl,
    timeout=timeout,
    verify_certs=verify_certs,
    connection_class=RequestsHttpConnection,
    pool_maxsize=pool_maxsize,
)

2. 构建查询体

定义您的查询条件。为了提高性能，建议使用fields参数只返回您需要的字段，并设置_source: False以避免返回整个_source文档。

query_body = {
    "size": 10000, # 每次滚动请求返回的最大文档数，通常设为10000
    "timeout": "300s", # 查询超时时间
    "query": {
        "bool": {
            "must": [
                {"match": {"type": "req"}}, # 匹配 'type' 字段为 'req' 的文档
                {"range": {"@timestamp": {"gte": "now-7d/d", "lte": "now/d"}}}, # 查询过去7天的数据
                {"wildcard": {"req_h_user_agent": {"value": "*googlebot*"}}}, # 用户代理包含 'googlebot'
            ]
        }
    },
    "fields": [ # 指定需要返回的字段
        "@timestamp",
        "resp_status",
        "resp_bytes",
        "req_h_referer",
        "req_h_user_agent",
        "req_h_host",
        "req_uri",
        "total_response_time",
    ],
    "_source": False, # 不返回完整的 _source 文档，只返回指定 fields
}

3. 执行滚动查询并处理结果

以下代码展示了如何发起初始滚动请求，然后循环迭代获取所有匹配的文档，并将它们写入CSV文件。

csv_file_path = "opensearch_report.csv" # 输出CSV文件名

try:
    # 1. 发起初始滚动查询
    # 'scroll' 参数指定了滚动上下文的有效期，例如 '1m' 表示1分钟
    print("正在发起初始滚动查询...")
    response = client.search(
        index="fastly-*", # 您的索引模式
        body=query_body,
        scroll='1m', # 保持滚动上下文1分钟
    )

    scroll_id = response.get("_scroll_id")
    if not scroll_id:
        print("没有找到匹配的结果或无法获取滚动ID。")
        exit()

    total_hits = response["hits"]["total"]["value"]
    print(f"查询到总计 {total_hits} 条匹配结果。")

    processed_hits_count = 0

    with open(csv_file_path, "w", newline='', encoding='utf-8') as f:
        writer = csv.writer(f)
        # 定义CSV文件头，与query_body中的 'fields' 对应
        writer.writerow([
            "timestamp", "url", "response_code", "bytes", "response_time",
            "referer", "user_agent"
        ])

        # 处理第一批结果
        hits = response["hits"]["hits"]
        for hit in hits:
            fields = hit.get("fields", {})
            # 注意：从 'fields' 获取的值通常是列表，需要取第一个元素 [0]
            writer.writerow([
                fields.get("@timestamp", [""])[0],
                fields.get("req_h_host", [""])[0] + fields.get("req_uri", [""])[0],
                fields.get("resp_status", [""])[0],
                fields.get("resp_bytes", [""])[0],
                fields.get("total_response_time", [""])[0],
                fields.get("req_h_referer", [""])[0],
                fields.get("req_h_user_agent", [""])[0],
            ])
            processed_hits_count += 1

        print(f"已处理 {processed_hits_count} 条结果。")

        # 2. 循环迭代获取后续结果
        while hits: # 当 hits 列表不为空时继续循环
            print(f"正在获取下一批结果 (已处理: {processed_hits_count}/{total_hits})...")
            response = client.scroll(
                scroll_id=scroll_id,
                scroll='1m', # 每次滚动请求都刷新滚动上下文的有效期
            )

            # 更新滚动ID，以防它发生变化 (虽然通常不会变)
            scroll_id = response.get("_scroll_id") 
            hits = response["hits"]["hits"]

            if not hits: # 如果没有更多结果，跳出循环
                break

            for hit in hits:
                fields = hit.get("fields", {})
                writer.writerow([
                    fields.get("@timestamp", [""])[0],
                    fields.get("req_h_host", [""])[0] + fields.get("req_uri", [""])[0],
                    fields.get("resp_status", [""])[0],
                    fields.get("resp_bytes", [""])[0],
                    fields.get("total_response_time", [""])[0],
                    fields.get("req_h_referer", [""])[0],
                    fields.get("req_h_user_agent", [""])[0],
                ])
                processed_hits_count += 1
            print(f"已处理 {processed_hits_count} 条结果。")

except Exception as e:
    print(f"在获取数据过程中发生错误: {e}")
finally:
    # 3. 清除滚动上下文 (可选但推荐)
    # 即使不手动清除，滚动上下文也会在有效期过后自动失效
    if scroll_id:
        try:
            client.clear_scroll(scroll_id=scroll_id)
            print(f"滚动上下文 {scroll_id} 已清除。")
        except Exception as e:
            print(f"清除滚动上下文失败: {e}")

print(f"所有匹配结果已检索完毕并保存到 {csv_file_path}。")

重要注意事项

1. 滚动上下文的生命周期： scroll参数（例如'1m'）定义了滚动上下文在OpenSearch服务器上保持活动的时长。每次调用client.scroll()时，这个计时器都会被重置。确保您的处理速度足够快，以免滚动上下文过期。
2. 资源消耗： 滚动查询会占用OpenSearch集群的内存和CPU资源，因为它需要维护搜索上下文。在处理超大规模数据时，应谨慎使用，并考虑集群的负载能力。
3. 清除滚动上下文： 虽然滚动上下文会在过期后自动清除，但为了及时释放资源，在完成数据检索后，强烈建议显式调用client.clear_scroll(scroll_id=scroll_id)来手动清除它。在上述代码的finally块中已经包含了这一操作。
4. 数据一致性： 滚动查询提供的是一个“时间点”快照。这意味着在滚动查询开始后，即使索引中的数据发生变化，滚动查询返回的结果集也不会受影响，它会返回查询开始那一刻的数据状态。
5. 替代方案：search_after： 对于需要实时或近实时数据，且每次查询只需要获取下一页数据而不是全部数据的场景，search_after API可能是更好的选择。它不维护搜索上下文，因此对资源消耗更小，但需要您手动管理上次查询的最后一个文档的排序值。然而，对于获取全量数据，Scroll API通常更简单直接。
6. 错误处理： 在实际应用中，务必添加健壮的错误处理机制，例如网络中断、OpenSearch集群故障等。

总结

通过opensearch-py库结合OpenSearch的Scroll API，您可以轻松地克服10,000条结果的限制，高效地检索和处理大规模数据集。理解其工作原理并正确实现迭代逻辑是关键。始终记住在完成操作后清除滚动上下文，以优化集群资源利用。

위 내용은 OpenSearch Python Client : Scroll API는 전체 데이터 검색을 구현합니다의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!