Elasticsearch Python脚本迭代多次后es.index操作卡住超时如何解决

问题根因 #

脚本卡住主要是以下几个原因共同导致的：

• 单文档逐次操作的请求量太大：每处理一个ID需要发起exists、get、index三次独立请求，处理700个文档就会产生超过2000次高频同步请求，很快就占满Elasticsearch的连接池，导致后续请求阻塞超时，而Colab环境的系统资源、ES单节点性能本身就有限，更容易触发瓶颈。
• 写放过大：你采用「读全量文档-本地修改-全量覆盖写入」的逻辑，随着关联关系越来越多，单文档体积越来越大，每次读写的传输、写入开销会持续升高，到阈值后直接卡住。
• 客户端配置不合理：你给单次index请求设置了极长的超时时间，请求阻塞后不会主动重试或抛出异常，只会无限等待直到底层连接超时。

优化方案 #

1. 优先本地预计算全量关联数据 #

不要读一行CSV就写一次ES，先把所有关联关系、出现次数在Python内存里计算完成，再统一写入ES，能最大程度减少和ES的交互次数。

2. 用批量写入接口替代单条写入 #

使用Elasticsearch客户端的helpers.bulk批量提交接口，一次提交几百条写入请求，请求量可以降到原来的几十分之一。

3. 优化索引写入阶段的配置 #

写入前临时调整索引配置降低额外开销，写入完成后再恢复：

• 把索引刷新间隔设为-1，关闭写入过程中的自动刷新
• 把索引副本数设为0，单节点ES不需要副本，写入完成后可按需调整

4. 修正客户端配置 #

初始化ES客户端时设置合理的超时、重试策略，避免无意义的长时间阻塞。

优化后代码示例 #

from elasticsearch import Elasticsearch, helpers
from collections import defaultdict

ES_NODES = "http://localhost:9200"
# 初始化客户端时配置合理的超时、重试策略
es = Elasticsearch(
    hosts=[ES_NODES],
    request_timeout=30,
    max_retries=3,
    retry_on_timeout=True
)

path = '/content/test.csv'
index_name = 'relacoes_produtos'

# 创建索引（如果不存在）
if not es.indices.exists(index=index_name):
    print('Criando índice')
    # 创建索引时临时配置为写入优化模式
    index_settings = {
        "settings": {
            "number_of_shards": 1,
            "number_of_replicas": 0,
            "refresh_interval": "-1"
        }
    }
    res = es.indices.create(index=index_name, body=index_settings)
    print("Response from server: {}".format(res))

# 第一步：本地预计算所有关联关系和出现次数
prod_stats = defaultdict(lambda: {"ocorrencias": 0, "relacoes": defaultdict(int)})

with open(path, 'r', encoding="utf-8") as f:
    for line in f:
        line = line.strip().replace(' ','').rstrip('\x00')
        produtos = list(dict.fromkeys(line.split(',')))
        # 跳过空行
        if not produtos:
            continue
        # 遍历当前行的所有ID更新计数
        for id in produtos:
            prod_stats[id]["ocorrencias"] += 1
            # 更新关联计数
            for other_id in produtos:
                if other_id != id:
                    prod_stats[id]["relacoes"][other_id] += 1

# 第二步：批量写入ES
actions = []
batch_size = 200  # 每200条提交一次，可根据实际情况调整

for prod_id, stats in prod_stats.items():
    action = {
        "_op_type": "index",
        "_index": index_name,
        "_id": str(prod_id),
        "_source": {
            "id": str(prod_id),
            "ocorrencias": stats["ocorrencias"],
            "relacoes": dict(stats["relacoes"])
        }
    }
    actions.append(action)
    # 攒够批量大小就提交
    if len(actions) >= batch_size:
        helpers.bulk(es, actions)
        actions = []
# 提交剩余的请求
if actions:
    helpers.bulk(es, actions)

# 写入完成后恢复索引正常配置
es.indices.put_settings(
    index=index_name,
    body={
        "refresh_interval": "1s",
        "number_of_replicas": 0  # 如果需要副本可自行修改为1
    }
)
# 手动触发一次刷新确保所有数据可见
es.indices.refresh(index=index_name)

内容的提问来源于stack exchange，提问作者anderici