欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

从Elasticsearch索引库中读取数据的操作

程序员文章站 2022-07-14 15:12:53
...

从ES索引库中读取数据的代码逻辑如下:

#!/usr/bing/env python
# -*-coding:utf-8-*-
# author:xx
# datetime:21-x-x 下午xx:xx
# software:PyCharm

import os
import sys
import json
import inspect

filename = inspect.getframeinfo(inspect.currentframe()).filename
matrix_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(filename))))
sys.path.insert(0, matrix_dir)

from datetime import datetime
from elasticsearch import Elasticsearch


class ReadData(object):

    def __init__(self):
        # connect es
        self.es_client = Elasticsearch(hosts="localhost:xxxx/")
        # index name (medical)
        self.index_name = "xx-xx-index"
        # index name (public chat)
        self.index_name = "ssk-public-index-1"
        # data path
        self.data_path = os.path.join(matrix_dir, "xxx/xxx/")
        # control
        self.debug = True

    # save data into file
    def save_data(self, data_list):
        # open file
        with open(os.path.join(self.data_path, "xxx_xxx.data"), 'w', encoding="utf-8") as file:
            # traverse the data
            for temp_dict in data_list:
                # transfer into json format
                temp_dict = json.dumps(temp_dict, ensure_ascii=False)
                file.write(temp_dict)
                file.write("\n")

        # prompt
        print("*******save finish*****")

    # read data from es
    def read_es(self):
        # set the container
        result_list = list()
        # 检查index是否存在
        if self.es_client.indices.exists(index=self.index_name):
            page = self.es_client.search(index=self.index_name, scroll="20m", size=200)
            # 游标用于输出es查询出的scroll id
            scroll_id = page['_scroll_id']
            if self.debug == True:
                print("the scroll_id is : %s" %(scroll_id))
                print("=*="*10)

            # es查询出的结果总量
            scroll_size = page['hits']['total']
            if self.debug == True:
                print("the total number is : %d" %(scroll_size))
                print("=*=" * 10)

            # traverse the data
            for source_dict in page['hits']['hits']:
                # get the data info
                temp_dict = source_dict["_source"]
                # add the data into list
                result_list.append(temp_dict)

            while scroll_size > 0:
                page = self.es_client.scroll(scroll_id=scroll_id, scroll="20m")
                scroll_id = page['_scroll_id']
                # get the page size
                scroll_size = len(page['hits']['hits'])
                if self.debug == True:
                    print("the scroll number is : %d" % (scroll_size))
                    print("=*=" * 10)

                # traverse the data
                for source_dict in page['hits']['hits']:
                    # get the data info
                    temp_dict = source_dict["_source"]
                    # add the data into list
                    result_list.append(temp_dict)

            if self.debug == True:
                print("*******read finish start save*******")
                print("=*=" * 10)
            # save data into file
            self.save_data(result_list)

        else:
            print('{} not find in es'.format(self.index_name))
            return None

if __name__ == "__main__":
    # the start_time
    start_time = datetime.now()
    # define the testObject
    dataObject = ReadData()
    # call the function
    dataObject.read_es()
    # the end_time
    end_time = datetime.now()
    use_time = end_time - start_time
    print("run time is:%ss||time:%s" % (use_time.seconds, use_time))