实用脚本能自动管理Elasticsearch吗？

wen 实用脚本 2026-07-21 4

本文目录导读：

索引生命周期管理脚本
集群健康监控脚本
索引优化与快照管理脚本
综合管理脚本（支持多操作）
使用方式
推荐工具 & 平台

是的，有很多实用脚本可以自动管理Elasticsearch,下面我为您整理几个常用场景的自动化脚本示例：

索引生命周期管理脚本

#!/bin/bash
# ES索引清理脚本 - 删除7天前的索引
ES_HOST="localhost:9200"
RETENTION_DAYS=7
# 获取所有索引
indices=$(curl -s "${ES_HOST}/_cat/indices?v" | awk '{print $3}' | grep -v "index")
# 计算截止日期
cutoff_date=$(date -d "${RETENTION_DAYS} days ago" +%Y.%m.%d)
for index in $indices; do
    # 提取索引中的日期（假设格式为 xxx-2023.01.01）
    index_date=$(echo $index | grep -oP '\d{4}\.\d{2}\.\d{2}')
    if [ -n "$index_date" ] && [[ "$index_date" < "$cutoff_date" ]]; then
        echo "删除索引: $index"
        curl -X DELETE "${ES_HOST}/${index}"
    fi
done

集群健康监控脚本

#!/usr/bin/env python3
# ES集群健康检查与自动恢复
import requests
import json
import smtplib
from datetime import datetime
ES_HOST = "http://localhost:9200"
ALERT_EMAIL = "admin@example.com"
def check_cluster_health():
    """检查集群健康状态"""
    response = requests.get(f"{ES_HOST}/_cluster/health")
    health = response.json()
    status = health['status']
    if status == 'red':
        # 自动尝试分配未分配的分片
        print(f"[{datetime.now()}] 集群状态为RED，尝试恢复...")
        retry_unassigned_shards()
        send_alert(f"集群状态异常: {status}")
    elif status == 'yellow':
        print(f"[{datetime.now()}] 警告：集群状态为YELLOW")
    return health
def retry_unassigned_shards():
    """重试分配未分配的分片"""
    data = {
        "commands": [{
            "allocate_stale_primary": {
                "index": "*",
                "shard": 0,
                "node": "node-1"
            }
        }]
    }
    requests.post(f"{ES_HOST}/_cluster/reroute", json=data)
def send_alert(message):
    """发送告警邮件"""
    # 邮件发送逻辑
    pass
if __name__ == "__main__":
    check_cluster_health()

索引优化与快照管理脚本

#!/bin/bash
# ES索引优化与快照管理
ES_HOST="localhost:9200"
SNAPSHOT_REPO="backup_repo"
DATE=$(date +%Y%m%d)
# 1. 强制合并小段（force merge）
echo "开始优化索引段..."
curl -X POST "${ES_HOST}/_all/_forcemerge?max_num_segments=5"
# 2. 创建快照
echo "创建快照..."
curl -X PUT "${ES_HOST}/_snapshot/${SNAPSHOT_REPO}/snapshot_${DATE}" -H 'Content-Type: application/json' -d'
{
  "indices": "important-*",
  "ignore_unavailable": true,
  "include_global_state": false
}'
# 3. 清理7天前的旧快照
echo "清理旧快照..."
OLD_DATE=$(date -d "7 days ago" +%Y%m%d)
curl -X DELETE "${ES_HOST}/_snapshot/${SNAPSHOT_REPO}/snapshot_${OLD_DATE}"
echo "操作完成！"

综合管理脚本（支持多操作）

#!/usr/bin/env python3
"""
ES全面管理自动化脚本
支持：索引管理、集群优化、备份恢复、监控告警
"""
import argparse
import requests
import json
import schedule
import time
from datetime import datetime, timedelta
class ESManager:
    def __init__(self, host):
        self.host = host
        self.base_url = f"http://{host}"
    def list_indices(self, pattern=None):
        """列出索引"""
        url = f"{self.base_url}/_cat/indices?v"
        if pattern:
            url += f"&index={pattern}"
        return requests.get(url).text
    def delete_old_indices(self, days=30, pattern="*"):
        """删除过期索引"""
        cutoff = datetime.now() - timedelta(days=days)
        response = requests.get(f"{self.base_url}/{pattern}/_settings")
        indices = response.json()
        for index_name in indices:
            # 解析索引创建时间（简化版）
            index_date_str = index_name.split('-')[-1]
            try:
                index_date = datetime.strptime(index_date_str, "%Y.%m.%d")
                if index_date < cutoff:
                    print(f"删除索引: {index_name}")
                    requests.delete(f"{self.base_url}/{index_name}")
            except ValueError:
                continue
    def optimize_cluster(self):
        """集群优化"""
        print("执行集群优化...")
        # 滚动重启（如果有需要）
        # 调整分片分配
        requests.put(f"{self.base_url}/_cluster/settings", json={
            "transient": {
                "cluster.routing.allocation.enable": "all"
            }
        })
        # 执行强制合并
        requests.post(f"{self.base_url}/_all/_forcemerge?max_num_segments=1")
    def check_and_alert(self):
        """健康检查并告警"""
        health = requests.get(f"{self.base_url}/_cluster/health").json()
        status = health['status']
        if status == 'red':
            self.send_alert(f"紧急：集群状态为RED，需要立即处理！")
        elif status == 'yellow':
            self.send_alert(f"警告：集群状态为YELLOW")
        return status
def main():
    parser = argparse.ArgumentParser(description='ES自动化管理工具')
    parser.add_argument('--host', default='localhost:9200', help='ES服务地址')
    parser.add_argument('--action', required=True, 
                      choices=['clean', 'optimize', 'backup', 'monitor', 'all'])
    parser.add_argument('--days', type=int, default=30, help='保留天数')
    parser.add_argument('--cron', action='store_true', help='定时执行模式')
    args = parser.parse_args()
    manager = ESManager(args.host)
    if args.cron:
        # 定时任务模式
        schedule.every().day.at("03:00").do(manager.delete_old_indices, args.days)
        schedule.every().week.do(manager.optimize_cluster)
        schedule.every().hour.do(manager.check_and_alert)
        while True:
            schedule.run_pending()
            time.sleep(60)
    else:
        # 单次执行模式
        if args.action in ['clean', 'all']:
            manager.delete_old_indices(args.days)
        if args.action in ['optimize', 'all']:
            manager.optimize_cluster()
        if args.action in ['monitor', 'all']:
            manager.check_and_alert()
if __name__ == "__main__":
    main()

使用方式

# 安装依赖
pip install requests schedule
# 清理30天前的索引
python es_manager.py --host localhost:9200 --action clean --days 30
# 全量优化并备份
python es_manager.py --host localhost:9200 --action all
# 定时任务模式（每天凌晨3点清理）
python es_manager.py --host localhost:9200 --action clean --cron

实用脚本能自动管理Elasticsearch吗？

索引生命周期管理脚本

集群健康监控脚本

索引优化与快照管理脚本

综合管理脚本（支持多操作）

使用方式

推荐工具 & 平台