Elasticsearch中执行多个_update_by_query的最快方法是什么？

常见思路的优缺点 #

• 暴力循环逐个执行：实现最简单但效率最低，每个请求都要单独建立HTTP连接、处理响应，延迟叠加，任务数量多时会大量浪费资源，拖慢整体速度。
• 查询后用BULK API更新：先通过查询获取目标文档的_id和_version，再构造BULK请求批量更新。这种方式比循环快，但多了一次查询开销，且如果查询到更新期间文档被修改，容易出现版本冲突。

最优执行方案 #

如果你的需求是执行多个逻辑不同的_update_by_query（比如不同查询条件、不同更新脚本），最优方式是异步并行执行任务+单个任务性能优化：

1. 异步并行提交任务

放弃单线程循环，用异步框架或线程池同时提交多个_update_by_query请求，并发数根据ES集群的硬件配置调整（建议每个数据节点维持2-4个并发更新任务，避免压垮集群）。

示例（Python异步实现）：

import aiohttp
import asyncio

async def execute_update(session, es_endpoint, update_dsl):
    async with session.post(f"{es_endpoint}/_update_by_query", json=update_dsl) as response:
        return await response.json()

async def main():
    es_endpoint = "http://your-es-cluster:9200"
    # 定义多个不同的_update_by_query任务
    update_tasks = [
        {
            "query": {"term": {"category": "book"}},
            "script": {"source": "ctx._source.status = 'processed'", "lang": "painless"},
            "retry_on_conflict": 3
        },
        {
            "query": {"range": {"price": {"gte": 100}}},
            "script": {"source": "ctx._source.discount = true", "lang": "painless"},
            "retry_on_conflict": 3
        }
        # 添加更多任务...
    ]

    async with aiohttp.ClientSession() as session:
        # 批量提交异步任务
        tasks = [execute_update(session, es_endpoint, dsl) for dsl in update_tasks]
        results = await asyncio.gather(*tasks)
        # 处理每个任务的结果
        for idx, res in enumerate(results):
            print(f"任务 {idx+1} 执行结果: {res}")

if __name__ == "__main__":
    asyncio.run(main())

2. 优化单个_update_by_query的性能

• 缩小查询范围：用精确的过滤条件（比如term、range）替代模糊匹配，避免全索引扫描，减少每个任务处理的文档量。
• 异步后台执行：添加wait_for_completion=false参数，让ES在后台执行任务，客户端立即获取任务ID，无需等待任务完成，能更快提交后续任务。
• 调整滚动批次：设置scroll_size参数（比如"scroll_size": 2000），控制每次批量处理的文档数，平衡内存占用和处理速度。
• 处理冲突：添加retry_on_conflict参数（比如"retry_on_conflict": 3），自动重试版本冲突的更新，避免任务失败。

3. 特殊场景：统一更新逻辑的批量处理

如果多个_update_by_query的更新逻辑完全一致，只是查询条件不同，可以合并查询条件，用单个_update_by_query处理所有文档，或者先查询出所有目标文档，再用BULK API批量更新，进一步减少请求次数。

注意事项 #

• 实时监控ES集群的CPU、内存、磁盘IO负载，根据负载动态调整并发数，避免集群过载。
• 大规模更新任务建议分批次执行，比如每次提交5-10个任务，完成一批后再提交下一批。
• 对于重要数据，执行更新前建议先备份索引，避免意外数据丢失。

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