本文目录导读：
我来介绍几种批量查询Elasticsearch的脚本方法：
使用curl脚本（最基础）
bash脚本 - 批量查询多个索引
#!/bin/bash
# 定义ES连接信息
ES_HOST="http://localhost:9200"
INDEXES=("index1" "index2" "index3")
# 查询所有索引
for index in "${INDEXES[@]}"; do
echo "=== Querying index: $index ==="
curl -s -X GET "$ES_HOST/$index/_search" -H 'Content-Type: application/json' -d '{
"query": {
"match_all": {}
},
"size": 10
}' | jq '.hits.hits'
echo -e "\n"
done
使用Python脚本（推荐）
安装依赖
pip install elasticsearch pandas
批量查询脚本
from elasticsearch import Elasticsearch
from elasticsearch.helpers import scan
import json
from datetime import datetime
import pandas as pd
class ESBatchQuery:
def __init__(self, hosts=['http://localhost:9200']):
self.es = Elasticsearch(hosts)
def query_by_ids(self, index, ids, batch_size=100):
"""
根据ID列表批量查询
"""
all_results = []
# 分批查询
for i in range(0, len(ids), batch_size):
batch_ids = ids[i:i+batch_size]
body = {
"query": {
"ids": {
"values": batch_ids
}
},
"size": len(batch_ids)
}
results = self.es.search(index=index, body=body)
all_results.extend([hit['_source'] for hit in results['hits']['hits']])
return all_results
def scroll_batch(self, index, query=None, scroll='2m', batch_size=1000):
"""
使用scroll API进行大批量数据导出
"""
if query is None:
query = {"query": {"match_all": {}}}
# 初始化scroll
response = self.es.search(
index=index,
body=query,
scroll=scroll,
size=batch_size
)
scroll_id = response['_scroll_id']
hits = response['hits']['hits']
total_docs = []
total_docs.extend([hit['_source'] for hit in hits])
# 持续滚动查询直到全部获取
while len(hits) > 0:
response = self.es.scroll(
scroll_id=scroll_id,
scroll=scroll
)
scroll_id = response['_scroll_id']
hits = response['hits']['hits']
total_docs.extend([hit['_source'] for hit in hits])
print(f"已获取: {len(total_docs)} 条记录")
# 清理scroll
self.es.clear_scroll(scroll_id=scroll_id)
return total_docs
def multi_index_query(self, indices, query, size=100):
"""
多索引查询
"""
results = {}
for index in indices:
try:
response = self.es.search(
index=index,
body=query,
size=size
)
results[index] = [hit['_source'] for hit in response['hits']['hits']]
print(f"{index}: 找到 {len(results[index])} 条记录")
except Exception as e:
print(f"{index}: 查询失败 - {str(e)}")
return results
def save_to_json(self, data, filename):
"""保存结果到JSON文件"""
with open(filename, 'w', encoding='utf-8') as f:
json.dump(data, f, ensure_ascii=False, indent=2)
def save_to_csv(self, data, filename):
"""保存结果到CSV文件"""
df = pd.DataFrame(data)
df.to_csv(filename, index=False, encoding='utf-8')
# 使用示例
if __name__ == "__main__":
# 初始化
es_query = ESBatchQuery(['http://localhost:9200'])
# 示例1: 按ID批量查询
ids = ['id1', 'id2', 'id3', 'id4', 'id5']
results = es_query.query_by_ids('my_index', ids)
# 示例2: 使用scroll导出全部数据
all_data = es_query.scroll_batch(
'my_index',
query={"query": {"range": {"timestamp": {"gte": "2024-01-01"}}}},
batch_size=5000
)
# 示例3: 多索引查询
indices = ['logs-2024-01', 'logs-2024-02', 'logs-2024-03']
query = {
"query": {
"bool": {
"must": [
{"match": {"status": "error"}},
{"range": {"response_time": {"gt": 1000}}}
]
}
}
}
multi_results = es_query.multi_index_query(indices, query)
# 保存结果
es_query.save_to_json(all_data, 'es_export.json')
es_query.save_to_csv(all_data, 'es_export.csv')
使用并发查询（高性能方案）
from concurrent.futures import ThreadPoolExecutor, as_completed
from elasticsearch import Elasticsearch
import threading
class ConcurrentESQuery:
def __init__(self, hosts):
self.local = threading.local()
self.hosts = hosts
def get_connection(self):
"""线程安全的连接管理"""
if not hasattr(self.local, 'es'):
self.local.es = Elasticsearch(self.hosts)
return self.local.es
def query_single_index(self, index_info):
"""查询单个索引"""
index, query = index_info
try:
es = self.get_connection()
results = es.search(index=index, body=query, size=1000)
return index, [hit['_source'] for hit in results['hits']['hits']]
except Exception as e:
return index, str(e)
def concurrent_multi_index_query(self, index_query_pairs, max_workers=5):
"""
并发查询多个索引
index_query_pairs: [(index1, query1), (index2, query2), ...]
"""
results = {}
with ThreadPoolExecutor(max_workers=max_workers) as executor:
future_to_index = {
executor.submit(self.query_single_index, pair): pair[0]
for pair in index_query_pairs
}
for future in as_completed(future_to_index):
index = future_to_index[future]
try:
index_name, data = future.result()
results[index_name] = data
print(f"完成查询: {index_name}, 获取 {len(data)} 条")
except Exception as e:
results[index] = f"失败: {str(e)}"
return results
# 使用示例
concurrent = ConcurrentESQuery(['http://localhost:9200'])
# 准备批量查询任务
tasks = [
('index1', {"query": {"match": {"field1": "value1"}}}),
('index2', {"query": {"range": {"price": {"gte": 100}}}}),
('index3', {"query": {"match_all": {}}}),
]
# 并发执行
batch_results = concurrent.concurrent_multi_index_query(tasks)
使用DSL查询构建器
from elasticsearch_dsl import Search, Q
class DSLBatchQuery:
def __init__(self, host='http://localhost:9200'):
self.client = connection.create(hosts=[host])
def complex_batch_query(self, index, conditions):
"""
构建复杂批量查询
conditions: [("field1", "value1"), ("field2", "value2")]
"""
s = Search(using=self.client, index=index)
# 构建多条件查询
queries = []
for field, value in conditions:
queries.append(Q('match', **{field: value}))
# 使用bool查询组合
s = s.query('bool', must=queries)
# 执行并返回结果
response = s.execute()
return [hit.to_dict() for hit in response]
# 使用示例
dsl_queries = [("status", "active"), ("category", "electronics")]
results = dsl_query.complex_batch_query("my_index", dsl_queries)
配置文件和主程序
# config.yaml
es:
hosts:
- http://localhost:9200
timeout: 30
max_retries: 3
queries:
- index: "logs-*"
query: {"match_all": {}}
output: "all_logs.json"
- index: "users"
query: {"match": {"status": "active"}}
output: "active_users.csv"
# main.py
import yaml
from es_batch import ESBatchQuery
def load_config(config_file='config.yaml'):
with open(config_file, 'r') as f:
return yaml.safe_load(f)
def run_batch_query(config):
es = ESBatchQuery(config['es']['hosts'])
for query_config in config['queries']:
print(f"查询索引: {query_config['index']}")
results = es.scroll_batch(
query_config['index'],
query_config['query']
)
if query_config['output'].endswith('.json'):
es.save_to_json(results, query_config['output'])
else:
es.save_to_csv(results, query_config['output'])
print(f"完成: {len(results)} 条记录")
if __name__ == "__main__":
config = load_config()
run_batch_query(config)
使用建议
- 数据量大时：使用scroll API而不是普通的search
- 需要高性能：使用并发查询（ThreadPoolExecutor）
- 简单查询：使用curl脚本快速测试
- 生产环境：使用Python脚本，添加错误处理和日志
这些脚本可以根据你的具体需求进行调整和扩展。
