从Elasticsearch索引库中读取数据的操作
从ES索引库中读取数据的代码逻辑如下:#!/usr/bing/env python# -*-coding:utf-8-*-# author:xx# datetime:21-x-x 下午xx:xx# software:PyCharmimport osimport sysimport jsonimport inspectfilename = inspect.getframeinfo(inspect.c
·
从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))
更多推荐
已为社区贡献1条内容
所有评论(0)