如何使用Python在ElasticSearch中批量执行多个查询

你之前调用msearch失败是因为没遵循Elasticsearch批量查询的格式要求，msearch的请求体需要是交替的元数据对象和查询对象组成的列表，而非直接传入查询字典列表。以下是正确实现步骤：

1. 构造符合格式的批量请求体 #

每个查询需要搭配一个元数据对象（用于指定索引、类型等），两者交替放入列表。如果所有查询都用同一个索引，元数据对象可简化为{}（但结构必须保留）。

示例代码：

# 假设query_list_all是你的查询字典列表
# 方式1：循环构造
msearch_body = []
for query in query_list_all:
    # 元数据：指定查询索引（若调用msearch时已指定全局index，可改为{}）
    msearch_body.append({"index": "index"})
    # 加入当前查询体
    msearch_body.append(query)

# 方式2：列表推导式简化写法
msearch_body = [item for q in query_list_all for item in [{"index": "index"}, q]]

2. 调用msearch发送批量查询 #

注意复用Elasticsearch连接（不要每次查询都新建连接，提升效率），然后传入构造好的请求体：

from elasticsearch import Elasticsearch

# 初始化连接，全局复用
es = Elasticsearch([uri])

# 发送批量查询，index参数为全局默认索引（若元数据已指定，可省略）
res = es.msearch(body=msearch_body, index="index")

# 处理返回结果：res['responses']是按顺序排列的每个查询的结果
for idx, result in enumerate(res["responses"]):
    print(f"第{idx+1}个查询结果: ", result)

关键说明 #

• 若单个查询需要覆盖全局设置（比如指定不同索引、自定义size），修改对应位置的元数据或查询体即可：

  # 某个查询指定不同索引+自定义返回条数
  msearch_body.append({"index": "another_index"})
  msearch_body.append({"query": {...}, "size": 200})
  

• 避免在循环中重复创建Elasticsearch实例，频繁建立连接会大幅降低性能。

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