Elasticsearch新手使用update_by_query更新50k文档遇连接超时问题求助

错误原因 #

你遇到的ConnectionTimeout是因为默认10秒的读取超时时间不足以完成5万条文档的更新操作，加上wait_for_completion=True会让客户端一直等待任务完成，超出超时阈值后就触发了错误。

解决方法及优化方案 #

1. 延长请求超时时间 #

直接在update_by_query调用中设置更长的超时时间，给任务足够的执行周期：

query = {
    "script": {
        "inline": "ctx._source.name='srujan'"
    },
    "query": {
        "match_all": {}
    }
}
response = client.update_by_query(
    body=query, 
    index=_index, 
    wait_for_completion=True,
    request_timeout=120  # 设置为120秒，可根据实际执行速度调整
)

2. 异步执行任务（推荐） #

将wait_for_completion设为False，客户端会立即返回任务ID，无需一直等待任务结束，从根源避免超时。后续可通过任务ID查询执行状态：

# 提交异步更新任务
response = client.update_by_query(
    body=query, 
    index=_index, 
    wait_for_completion=False
)
task_id = response['task']

# 后续查询任务执行状态
task_status = client.tasks.get(task_id=task_id)
print(task_status)

3. 分片并行处理 #

使用slice参数把更新任务拆分成多个子任务并行执行，降低单任务负载，提升处理效率：

# 分成4个分片并行，可根据集群节点数调整数量
query = {
    "script": {
        "inline": "ctx._source.name='srujan'"
    },
    "query": {
        "match_all": {}
    },
    "slice": {
        "id": 0,
        "max": 4
    }
}
# 循环提交所有分片任务
for i in range(4):
    query['slice']['id'] = i
    client.update_by_query(
        body=query, 
        index=_index, 
        wait_for_completion=False
    )

4. 分批增量更新 #

如果上述方案仍有压力，可结合scroll API分批获取文档ID，再批量更新，灵活控制每批处理的文档数量：

# 滚动查询获取文档ID
scroll = client.search(
    index=_index,
    body={"query": {"match_all": {}}},
    scroll='2m',  # 滚动上下文保留时间
    size=1000  # 每批获取1000条
)
scroll_id = scroll['_scroll_id']
total_docs = scroll['hits']['total']['value']

while total_docs > 0:
    # 提取当前批次的文档ID
    doc_ids = [hit['_id'] for hit in scroll['hits']['hits']]
    # 构造批量更新请求体
    bulk_body = []
    for doc_id in doc_ids:
        bulk_body.append({
            "update": {
                "_index": _index,
                "_id": doc_id
            }
        })
        bulk_body.append({
            "script": {
                "inline": "ctx._source.name='srujan'"
            }
        })
    # 执行批量更新
    client.bulk(body=bulk_body)
    # 滚动获取下一批文档
    scroll = client.scroll(scroll_id=scroll_id, scroll='2m')
    scroll_id = scroll['_scroll_id']
    total_docs = len(scroll['hits']['hits'])

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