从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))
上一篇: 使用idea, sparksql读取hive中的数据
下一篇: Handler消息传递机制
推荐阅读
-
SQL Server 数据库调整表中列的顺序操作方法及遇到问题
-
phpwind中的数据库操作类
-
php中CI操作多个数据库的代码
-
Spring jdbc中数据库操作对象化模型的实例详解
-
从数据库中读取一个图片并保存为一个图片文件
-
面试|简单描述MySQL中,索引,主键,唯一索引,联合索引 的区别,对数据库的性能有什么影响(从读写两方面)
-
Oracle数据库中emp表的基本操作实例讲解
-
PHP 在5.1.* 和5.2.*之间 PDO数据库操作中的不同之处小结
-
VS中C#读取app.config数据库配置字符串的三种方法
-
我们在删除SQL Sever某个数据库表中数据的时候,希望ID重新从1开始,而不是紧跟着最后一个ID开始需要的命令