使用python获取指定时间段的elasticsearch数据

本脚本已经在生产环境完成校验，涉及到的知识点已经在脚本添加注释。

#!/usr/bin/evn python3
# -*-coding:utf-8-*-
# Author: zhaoyuanji
# Filename: get_data_from_es.py

import pandas as pd
from opensearchpy import OpenSearch
import warnings

warnings.filterwarnings('ignore')

class get_gateway_not_cloud:

    def __init__(self, es_ip, username, password):

        self.es_ip = es_ip
        self.username = username
        self.password = password

        self.es = OpenSearch(
            self.es_ip,
            verify_certs = False,
            verify_hostname = False,
            http_auth=(self.username, self.password),
            scheme="https",
            port=9200,
            timeout = 300
        )
        self.save_data = []

    def get_data(self,start,end):

        data = self.es.search(
                index='logstash-*',
                body={
                  "size": 1000000,
                  "track_total_hits": True,
                  "query": {
                    "bool": {
                      "filter": [
                        {
                          "match_all": {
                          }
                        },
                        {
                          "match_phrase": {
                            "appname": "apptest"
                          }
                        },
                        {
                        "bool": {
                            "should": [
                              {
                                "match_phrase": {
                                  "message": "10.1.10.3:8080"
                                }
                              },
                              {
                                "match_phrase": {
                                  "message": "10.1.10.2:8080"
                                }
                              }
                            ]
                          }
                        },
                        {
                          "range": {
                            "@timestamp": {
                                      "format": "strict_date_optional_time",
                                          "gte": start + 'T16:00:00.000Z',  # UTC时间从东八区前一天的16:00:00开始
                                          "lte": end + 'T15:59:59.999Z'     # UTC时间到东八区的15:59:59结束
                                  }
                                  }
                                }
                              ]
                            }
                        }
                }
        )
        print("查询结果一共有{0}条数据，包含重复数据。".format(len(data['hits']['hits'])))
        print("获取中......")
        for d in data['hits']['hits']:
            msg = d['_source']['message']

            json_data = {
                'appid': msg.split()[0].split(':')[1],
                'clientip': msg.split()[1].split(':')[1],
            }

            self.save_data.append(json_data)

        data = pd.DataFrame(self.save_data)
        data = data[["appid", "clientip"]]
        data = data.groupby(["appid", "clientip"]).size().reset_index(name='计数')
        data = data.sort_values(by="计数")

        data.to_excel(start + '_' + end + '_apptest_result.xlsx', sheet_name='统计结果', index=False)

        print(start + '_' + end + '_apptest_result.xlsx has been generated successfully.')


if __name__ == '__main__':
    obj = get_gateway_not_cloud('10.10.1.4', 'kibana', 'ELK@123')
    obj.get_data("2024-10-17", "2024-10-18")

复制