使用Python脚本进行es数据清理

时间 2020-10-24

标签 python json 服务器 app 运维 curl ide 函数测试 url 栏目 Python 繁體版

原文原文链接

需求背景

业务系统将各种的报表和统计数据存放于ES中，因为历史缘由，系统天天均以全量方式进行统计，随着时间的推移，ES的数据存储空间压力巨大。同时因为没有规划好es的索引使用，个别索引甚至出现超过最大文档数限制的问题，现实状况给运维人员带来的挑战是须要以最小的代价来解决这个问题。下面之内网开发、测试环境举例使用python脚本解决这个问题。python

Each Elasticsearch shard is a Lucene index. There is a maximum number of documents you can have in a single Lucene index. As of LUCENE-5843, the limit is 2,147,483,519 (= Integer.MAX_VALUE - 128) documents. You can monitor shard sizes using the _cat/shards API.json

实现思路

es自己支持“_delete_by_query”的形式对查询出来的数据进行删除。首先咱们经过”_cat/indices“入口获取当前es服务上全部的索引信息。服务器

第一列表示索引当前的健康状态
第三列表示索引的名称
第四列表示索引在服务器上的存储目录名
第5、六列表示索引的副本数和分片信息
第七列表示当前索引的文档数
最后两列分别表示当前索引的存储占用空间，倒数第二列等于倒数第一列乘以副本数app

其次咱们经过curl形式拼接成删除命令发送给es服务端执行，其中createtime字段为数据的产生时间，单位为毫秒运维

curl -X POST "http://192.168.1.19:9400/fjhb-surveyor-v2/_delete_by_query?pretty" -H 'Content-Type: application/json' -d '
     {"query":{ "range": {
            "createTime": {   
                "lt": 1580400000000,    
                "format": "epoch_millis"
            }
        }
}}'

具体实现

#!/usr/bin/python
# -*- coding: UTF-8 -*-

###导入必须的模块
import requests
import time
import datetime
import os

#定义获取ES数据字典函数，返回索引名和索引占用存储空间大小字典
def getData(env):
    header = {"Content-Type": "application/x-www-form-urlencoded",
              "user-agent": "User-Agent:Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/62.0.3202.94 Safari/537.36"
}
    data = {}
    with open('result.txt','w+') as f:
        req = requests.get(url=env+'/_cat/indices',headers=header).text
        f.write(req)
        f.seek(0)
        for line in f.readlines():
            data[line.split()[2]] = line.split()[-1]
    return data

#定义unix时间转换函数，以毫秒形式返回，返回值为int类型
def unixTime(day):
    today = datetime.date.today()
    target_day = today + datetime.timedelta(day)
    unixtime = int(time.mktime(target_day.timetuple())) * 1000
    return unixtime

#定义删除es数据函数，调用系统curl命令进行删除，须要传入环境、须要删除数据的时间范围（即多少天以前的数据）参数，因为索引数量众多，咱们只处理超过1G的索引便可
def delData(env,day):
    header = 'Content-Type: application/json'
    for key, value in getData(env).items():
        if 'gb' in value:
            size = float(value.split('gb')[0])
            if size > 1:
                url = env + '/' + key + '/_delete_by_query?pretty'
                command = ("curl -X POST \"%s\" -H '%s' "
                           "-d '{\"query\":{ \"range\": {\"createTime\": {\"lt\": %s,\"format\": \"epoch_millis\"}}}}'" % (
                           url, header, day))
                print(command)
                os.system(command)

if __name__ == '__main__':
    dev = 'http://192.168.1.19:9400'
    test1 = 'http://192.168.1.19:9200'
    test2 = 'http://192.168.1.19:9600'
    day = unixTime(-30)
    delData(dev,day)
    delData(test1,day)
    delData(test2,day)

结果验证

删除前

删除后
curl

注意事项

一、目前脚本采用操做系统crontab方式调度，一天运行一次
二、首次删除由于数据量庞大，须要耗费较长时间，后续天天删除一天的数据量，删除效率尚可
三、脚本未考虑服务器报错等例外状况与告警通知，实际应用场景须要进行补充
四、脚本未考虑日志记录，实际应用场景须要进行补充ide