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

从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))