Elasticsearch 的 scroll API

对于大量数据，可以使用 Elasticsearch 的 scroll API 来分批次地读取数据，以避免一次性读取所有数据造成的内存负担。这段代码使用滚动查询（scroll）来分批次地读取数据。首先，它发送初始的搜索请求，并获取第一批数据。然后，使用滚动查询持续获取剩余的数据，直到所有数据都被读取完毕。最后，记得清除滚动查询的状态，释放相关资源。以下是使用 scroll API 的示例代码：

from elasticsearch import Elasticsearch

def scroll_query(es, index_name, query, scroll_size=1000, scroll_time='1m'):
    """
    执行滚动查询，逐批获取数据并处理结果

    Args:
    - es: Elasticsearch 实例
    - index_name: 要查询的索引名称
    - query: 查询语句
    - scroll_size: 每次滚动查询获取的文档数量，默认为 1000
    - scroll_time: 滚动查询的保持时间，默认为 '1m'

    Returns:
    - None
    """
    # 初始化滚动查询
    result = es.search(index=index_name, body=query, size=scroll_size, scroll=scroll_time)
    scroll_id = result['_scroll_id']
    total_docs = result['hits']['total']['value']

    # 处理第一批数据
    hits = result['hits']['hits']
    for hit in hits:
        source_data = hit['_source']
        print(source_data)  # 处理你的数据，这里简单打印出来作为示例

    # 继续滚动查询获取剩余数据
    while len(hits) > 0:
        result = es.scroll(scroll_id=scroll_id, scroll=scroll_time)
        hits = result['hits']['hits']
        total_docs += result['hits']['total']['value']
        for hit in hits:
            source_data = hit['_source']
            print(source_data)  # 处理你的数据，这里简单打印出来作为示例
            
    print(total_docs) #   # 数据总数
    # 清除滚动查询
    es.clear_scroll(scroll_id=scroll_id)

# 连接到Elasticsearch集群
es = Elasticsearch([{'host': 'localhost', 'port': 9200}])

# 确保ES实例可用
if es.ping():
    print("Connected to Elasticsearch")
else:
    print("Could not connect to Elasticsearch")

# 定义查询语句
query = {
    "query": {
        "match_all": {}  # 匹配所有文档
    }
}

# 调用滚动查询函数
scroll_query(es, 'your_index_name', query, scroll_size=1000, scroll_time='1m')

在这个函数中，我们将滚动查询的逻辑封装在 scroll_query 函数中，可以通过传入 Elasticsearch 实例、索引名称、查询语句以及其他参数来执行滚动查询。你可以根据需要调整 scroll_size 和 scroll_time 参数来控制每次查询的文档数量和滚动查询的保持时间。调用该函数后，会依次输出查询到的数据。