如何用脚本批量查询Elasticsearch？

本文目录导读：

1. 使用curl脚本（最基础）
2. 使用Python脚本（推荐）
3. 使用并发查询（高性能方案）
4. 使用DSL查询构建器
5. 配置文件和主程序
6. 使用建议

我来介绍几种批量查询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)

使用建议

1. 数据量大时：使用scroll API而不是普通的search
2. 需要高性能：使用并发查询（ThreadPoolExecutor）
3. 简单查询：使用curl脚本快速测试
4. 生产环境：使用Python脚本，添加错误处理和日志

这些脚本可以根据你的具体需求进行调整和扩展。