Elasticsearch 8.12.1中高效获取10万+全量记录并缓存为字典的实现方案咨询

嗨，针对你这个在Elasticsearch 8.12.1里批量拉取10万+数据并缓存成字典的需求，我来给你几个实用的方案和代码示例，完全适配你的ES版本~

一、官方推荐方案：Point in Time (PIT) + Search After #

ES 8.x之后官方已经不推荐使用Scroll API了，取而代之的是Point in Time (PIT) + Search After的组合，它比Scroll更高效，还能避免Scroll长期占用集群资源的问题，非常适合你这种一次性全量拉取的场景。

实现思路 #

1. 先创建一个PIT（时间点快照），保留当前索引的状态，避免拉取过程中数据变化导致的不一致
2. 第一次查询获取第一批数据，同时记录下最后一条数据的排序值（用于后续的search_after参数）
3. 循环调用查询，每次用上一批最后一条的排序值作为search_after的入参，直到拉完所有数据
4. 最后关闭PIT释放资源，同时把拉取到的数据整理成{IP: ISP名称}的字典

完整Python代码 #

首先确保你安装了对应版本的ES Python客户端：

pip install elasticsearch==8.12.1

然后是核心实现代码：

from elasticsearch import Elasticsearch
from elasticsearch.exceptions import ElasticsearchException

# 初始化ES客户端（根据你的集群认证方式调整，比如用户名密码/API Key）
es = Elasticsearch(
    "http://你的ES集群地址:9200",
    # 示例：用API Key认证
    api_key="你的API Key",
    # 示例：用用户名密码认证
    # basic_auth=("username", "password")
)

def fetch_and_cache_isp_records(index_name="isp_names"):
    ip_isp_cache = {}
    # 设置PIT保留时间，1分钟足够拉完10万条数据（可根据实际情况调整）
    pit_keep_alive = "1m"
    # 每次拉取的批次大小，建议设置为500-2000之间，平衡请求次数和ES压力
    batch_size = 1000

    try:
        # 1. 创建时间点快照（PIT）
        pit_response = es.open_point_in_time(index=index_name, keep_alive=pit_keep_alive)
        pit_id = pit_response["id"]

        # 2. 第一次查询，基于_id排序（_id唯一且稳定，无需额外创建排序字段）
        initial_response = es.search(
            pit={"id": pit_id, "keep_alive": pit_keep_alive},
            size=batch_size,
            sort=[{"_id": "asc"}],
            source=True  # 我们需要获取_source里的ISP名称
        )

        # 处理第一批数据
        for hit in initial_response["hits"]["hits"]:
            ip = hit["_id"]
            # 每条_source只有一个键值对（IP: ISP名称），直接取第一个值即可
            isp_name = next(iter(hit["_source"].values())).strip()
            ip_isp_cache[ip] = isp_name

        # 获取第一批最后一条数据的排序值，用于后续的search_after
        last_sort = initial_response["hits"]["hits"][-1]["sort"] if initial_response["hits"]["hits"] else None

        # 3. 循环拉取剩余数据
        while last_sort:
            response = es.search(
                pit={"id": pit_id, "keep_alive": pit_keep_alive},
                size=batch_size,
                sort=[{"_id": "asc"}],
                search_after=last_sort,
                source=True
            )

            batch_hits = response["hits"]["hits"]
            if not batch_hits:
                break

            # 处理当前批次数据
            for hit in batch_hits:
                ip = hit["_id"]
                isp_name = next(iter(hit["_source"].values())).strip()
                ip_isp_cache[ip] = isp_name

            # 更新last_sort为当前批次最后一条数据的排序值
            last_sort = batch_hits[-1]["sort"]

        # 4. 关闭PIT，释放ES资源（非常重要！）
        es.close_point_in_time(id=pit_id)
        print(f"成功拉取并缓存了{len(ip_isp_cache)}条ISP记录")
        return ip_isp_cache

    except ElasticsearchException as e:
        print(f"拉取ES数据时出现错误: {str(e)}")
        # 出错时尽量关闭PIT，避免资源泄漏
        if 'pit_id' in locals():
            try:
                es.close_point_in_time(id=pit_id)
            except:
                pass
        return {}

# 服务启动时调用，缓存到全局变量
isp_cache = fetch_and_cache_isp_records()

二、备选方案：Scroll API（一次性拉取可使用） #

虽然官方不推荐Scroll API，但如果是服务启动时的一次性全量拉取，其实也可以用（毕竟10万条数据量不算特别大）。不过还是优先推荐上面的PIT+Search After方案。

核心Python代码 #

from elasticsearch import Elasticsearch
from elasticsearch.exceptions import ElasticsearchException

es = Elasticsearch(
    "http://你的ES集群地址:9200",
    api_key="你的API Key"
)

def fetch_with_scroll(index_name="isp_names"):
    ip_isp_cache = {}
    scroll_time = "1m"
    batch_size = 1000

    try:
        # 1. 初始化Scroll查询
        initial_response = es.search(
            index=index_name,
            size=batch_size,
            scroll=scroll_time,
            query={"match_all": {}}
        )

        scroll_id = initial_response["_scroll_id"]
        total_hits = initial_response["hits"]["total"]["value"]
        print(f"检测到总共有{total_hits}条记录需要拉取")

        # 处理第一批数据
        for hit in initial_response["hits"]["hits"]:
            ip = hit["_id"]
            isp_name = next(iter(hit["_source"].values())).strip()
            ip_isp_cache[ip] = isp_name

        # 2. 循环拉取剩余数据
        while len(initial_response["hits"]["hits"]) > 0:
            initial_response = es.scroll(
                scroll_id=scroll_id,
                scroll=scroll_time
            )

            for hit in initial_response["hits"]["hits"]:
                ip = hit["_id"]
                isp_name = next(iter(hit["_source"].values())).strip()
                ip_isp_cache[ip] = isp_name

            # 更新scroll_id
            scroll_id = initial_response["_scroll_id"]

        # 3. 清除Scroll会话，释放资源
        es.clear_scroll(scroll_id=scroll_id)
        print(f"成功拉取并缓存了{len(ip_isp_cache)}条ISP记录")
        return ip_isp_cache

    except ElasticsearchException as e:
        print(f"拉取ES数据时出现错误: {str(e)}")
        if 'scroll_id' in locals():
            try:
                es.clear_scroll(scroll_id=scroll_id)
            except:
                pass
        return {}

# 服务启动时调用
isp_cache = fetch_with_scroll()

三、关键注意事项 #

1. 客户端版本匹配：确保你安装的elasticsearch Python客户端版本和你的ES集群版本完全一致（8.12.1），避免兼容性问题
2. 资源清理：不管用哪种方案，一定要记得清理PIT或Scroll会话，避免长期占用ES集群资源
3. 批次大小调整：batch_size可以根据你的ES集群性能调整，太大可能导致ES节点压力过大，太小会增加请求次数，建议在500-2000之间测试
4. 内存占用：10万条IP+ISP名称的字典，内存占用大概在几十MB左右，完全不用担心内存问题

备注：内容来源于stack exchange，提问作者arpit joshi