1.

# -*- coding: utf-8 -*-

import time
from elasticsearch import Elasticsearch
import  datetime


def create_snapshot(es):
    # 儲存時間戳計算
    # now
    nowtime = datetime.datetime.now()
    print(nowtime)
    now_timeStamp = int(time.mktime(nowtime.timetuple()))
    print(now_timeStamp)

    # indices_list = [".kibana_1",".kibana_task_manager"]
 

    indices_list =[]
    indices_string = ','.join(indices_list)
    index_body = {"indices": indices_string}
    #快照
    es.snapshot.create(repository="s3_repository_api", snapshot="mybackup-%s" % (now_timeStamp), body=index_body)



def delete_shapshot(es):
    nowtime = datetime.datetime.now()
    
 
print(nowtime)
    now_timeStamp = int(time.mktime(nowtime.timetuple()))
    #獲取快照
    get_snapshot_all=es.snapshot.get(repository="s3_repository_api",snapshot="_all")
    # es.snapshot.delete(repository="s3_repository_api",snapshot="mybackup-1653293680") #  1653293667   1653293680
    for v in get_snapshot_all.values():
        
 
for snapshot_name in v:
            snapshot_name=snapshot_name["snapshot"]
            if ":"  in snapshot_name:
                timeStamp=int(snapshot_name.split(":")[-1])
                if now_timeStamp - timeStamp > 604800:
                    es.snapshot.delete(repository="s3_repository_api", snapshot=snapshot_name)




def main():
    es = Elasticsearch(["192.168.118.133"],Transport=9200)
    create_snapshot(es)
    delete_shapshot(es)


if __name__ == '__main__':
    main()

定時任務：

crontab -l

0 2 * * * python3 /data/snapshot.py

參考：Elasticsearch API Reference — Python Elasticsearch client 8.2.0 documentation (elasticsearch-py.readthedocs.io)

使用python和snapshot備份ElasticSearch索引資料_jacobxian的部落格-CSDN部落格