使用Python對ElasticSearch獲取資料及操作

Version

Python ：2.7

ElasticSearch：6.3

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
    @Time    : 2018/7/4
    @Author  : LiuXueWen
    @Site    : 
    @File    : ElasticSearchOperation.py
    @Software: PyCharm
    @Description: 對elasticsearch資料的操作，包括獲取資料，傳送資料
"""
 

import elasticsearch
import json

import Util_Ini_Operation

class elasticsearch_data():
    def __init__(self,hosts,username,password,maxsize,is_ssl):
        # 初始化ini操作指令碼，獲取配置檔案
        try:
            # 判斷請求方式是否ssl加密
            if is_ssl == "true":
                # 獲取證書地址
                cert_pem = Util_Ini_Operation.get_ini("config.ini"
 
).get_key_value("certs","certs")
                es_ssl = elasticsearch.Elasticsearch(
                    # 地址
                    hosts=hosts,
                    # 使用者名稱密碼
                    http_auth=(username,password),
                    # 開啟ssl
                    use_ssl=True,
                    # 確認有加密證書
 

                    verify_certs=True,
                    # 對應的加密證書地址
                    client_cert=cert_pem
                )
                self.es = es_ssl
            elif is_ssl == "false":
                # 建立普通型別的ES客戶端
                es_ordinary = elasticsearch.Elasticsearch(hosts, http_auth=(username, password), maxsize=int(maxsize))
                self.es = es_ordinary
        except Exception as e:
            print(e)


    def query_data(self,keywords_list,date):
        gte = "now-"+str(date)
        query_data = {
            # 查詢語句
            "query": {
                "bool": {
                    "must": [
                        {
                            "query_string": {
                                "query": keywords_list,
                                "analyze_wildcard": True
                            }
                        },
                        {
                            "range": {
                                "@timestamp": {
                                    "gte": gte,
                                    "lte": "now",
                                    "format": "epoch_millis"
                                }
                            }
                        }
                    ],
                    "must_not": []
                }
            }
        }
        return query_data

    # 從es獲取資料
    def get_datas_by_query(self,index_name,keywords,param,date):
        '''
        :param index_name: 索引名稱
        :param keywords: 關鍵字詞，陣列
        :param param: 需要資料條件，例如_source
        :param date: 過去時間範圍,字串格式，例如過去30分鐘內資料，"30m"
        :return: all_datas 返回查詢到的所有資料（已經過param過濾）
        '''

        all_datas = []
        # 遍歷所有的查詢條件
        for keywords_list in keywords:
            # DSL語句
            query_data = self.query_data(keywords_list,date)
            res = self.es.search(
                index=index_name,
                body=query_data
            )
            for hit in res['hits']['hits']:
                # 獲取指定的內容
                response = hit[param]
                # 新增所有資料到資料集中
                all_datas.append(response)
        # 返回所有資料內容
        return all_datas

    # 當索引不存在建立索引
    def create_index(self,index_name):
        '''
        :param index_name: 索引名稱
        :return:如果建立成功返回建立結果資訊，試過已經存在建立新的index失敗返回index的名稱
        '''
        # 獲取索引的對映
        # index_mapping = IndexMapping.index_mapping
        # # 判斷索引是否存在
        # if self.es.indices.exists(index=index_name) is not True:
        #     # 建立索引
        #     res = self.es.indices.create(index=index_name,body=index_mapping)
        #     # 返回結果
        #     return res
        # else:
        #     # 返回索引名稱
        #     return index_name
        pass

    # 插入指定的單條資料內容
    def insert_single_data(self,index_name,doc_type,data):
        '''
        :param index_name: 索引名稱
        :param doc_type: 文件型別
        :param data: 需要插入的資料內容
        :return: 執行結果
        '''
        res = self.es.index(index=index_name,doc_type=doc_type,body=data)
        return res

    # 向ES中新增資料,批量插入
    def insert_datas(self,index_name):
        '''
        :desc 通過讀取指定的檔案內容獲取需要插入的資料集
        :param index_name: 索引名稱
        :return: 插入成功的資料條數
        '''
        insert_datas = []
        # 判斷插入資料的索引是否存在
        self.createIndex(index_name=index_name)
        # 獲取插入資料的檔案地址
        data_file_path = self.ini.get_key_value("datafile","datafilepath")
        # 獲取需要插入的資料集
        with open(data_file_path,"r+") as data_file:
            # 獲取檔案所有資料
            data_lines = data_file.readlines()
            for data_line in data_lines:
                # string to json
                data_line = json.loads(data_line)
                insert_datas.append(data_line)
        # 批量處理
        res = self.es.bulk(index=index_name,body=insert_datas,raise_on_error=True)
        return res

    # 從ES中在指定的索引中刪除指定資料（根據id判斷）
    def delete_data_by_id(self,index_name,doc_type,id):
        '''
        :param index_name: 索引名稱
        :param index_type: 文件型別
        :param id: 唯一標識id
        :return: 刪除結果資訊
        '''
        res = self.es.delete(index=index_name,doc_type=doc_type,id=id)
        return res

    # 根據條件刪除資料
    def delete_data_by_query(self,index_name,doc_type,param,gt_time,lt_time):
        '''
        :param index_name:索引名稱，為空查詢所有索引
        :param doc_type:文件型別，為空查詢所有文件型別
        :param param:過濾條件值
        :param gt_time:時間範圍，大於該時間
        :param lt_time:時間範圍，小於該時間
        :return:執行條件刪除後的結果資訊
        '''
        # DSL語句
        query_data = {
            # 查詢語句
            "query": {
                "bool": {
                    "must": [
                        {
                            "query_string": {
                                "query": param,
                                "analyze_wildcard": True
                            }
                        },
                        {
                            "range": {
                                "@timestamp": {
                                    "gte": gt_time,
                                    "lte": lt_time,
                                    "format": "epoch_millis"
                                }
                            }
                        }
                    ],
                    "must_not": []
                }
            }
        }
        res = self.es.delete_by_query(index=index_name,doc_type=doc_type,body=query_data,_source=True)
        return res

    # 指定index中刪除指定時間段內的全部資料
    def delete_all_datas(self,index_name,doc_type,gt_time,lt_time):
        '''
        :param index_name:索引名稱，為空查詢所有索引
        :param doc_type:文件型別，為空查詢所有文件型別
        :param gt_time:時間範圍，大於該時間
        :param lt_time:時間範圍，小於該時間
        :return:執行條件刪除後的結果資訊
        '''
        # DSL語句
        query_data = {
            # 查詢語句
            "query": {
                "bool": {
                    "must": [
                        {
                            "match_all": {}
                        },
                        {
                            "range": {
                                "@timestamp": {
                                    "gte": gt_time,
                                    "lte": lt_time,
                                    "format": "epoch_millis"
                                }
                            }
                        }
                    ],
                    "must_not": []
                }
            }
        }
        res = self.es.delete_by_query(index=index_name, doc_type=doc_type, body=query_data, _source=True)
        return res

    # 修改ES中指定的資料
    def update_data_by_id(self,index_name,doc_type,id,data):
        '''
        :param index_name: 索引名稱
        :param doc_type: 文件型別，為空表示所有型別
        :param id: 文件唯一標識編號
        :param data: 更新的資料
        :return: 更新結果資訊
        '''
        res = self.es.update(index=index_name,doc_type=doc_type,id=id,body=data)
        return res