OpenSearch Python客户端：Scroll API实现全量数据检索

pip install opensearch-py

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,
)

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
}

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}。")