Python 调用 ES、Solr、Phoenix的示例代码
程序员文章站
2024-01-03 20:15:22
#!/usr/bin/env python# -*- coding:utf-8 -*-# *************************************# @time : 2019/8/...
#!/usr/bin/env python # -*- coding:utf-8 -*- # ************************************* # @time : 2019/8/12 # @author : zhang fan # @desc : library # @file : mydatabases.py # @update : 2019/8/23 # ************************************* import elasticsearch import phoenixdb import pysolr import pymysql class myels(object): """ =================================================================== ===================== myels ========================= =================================================================== """ def __init__(self): self.els_conn = none def connect_to_els(self, host, port): """ 连接到elasticsearch服务器. """ self.els_conn = elasticsearch.elasticsearch([{'host': host, 'port': port}]) print('executing : connect to elastic search | %s' % self.els_conn) def get_els_data(self, query, index): """ 获取elasticsearch数据 """ print('executing : search | %s' % query) try: rst = self.els_conn.search(index=index, q=query) return rst['hits'] except exception as e: print('elastic search error | %s' % e) raise exception(e) class myphoenix(object): """ =================================================================== ===================== myphoenix ====================== =================================================================== """ def __init__(self): self.phoenix_conn = none self.phoenix_cursor = none def connect_to_phoenix(self, host, port=8765): """ 连接到phoenix服务器 """ address = 'http://{0}:{1}/'.format(host, port) print('executing : connect to phoenix | %s' % address) self.phoenix_conn = phoenixdb.connect(address, autocommit=true) self.phoenix_cursor = self.phoenix_conn.cursor() def set_schema(self, sql, schema): """ 设置schema """ pre_sub, sub, fol_sub = sql.upper().partition('from') fol_sub = ' ' + schema + '.' + fol_sub.strip() new_sql = ''.join([pre_sub, sub, fol_sub]) return new_sql def execute_phoenix_sql(self, sql): """ 执行sql语句 """ # sql = self.set_schema(sql, schema) print('executing : execute | %s' % sql) self.phoenix_cursor.execute(sql) def get_from_phoenix(self, sql): """ 获取phoenix数据 """ # sql = self.set_schema(sql, schema) print('executing : query | %s' % sql) try: self.phoenix_cursor.execute(sql) except exception as e: print('phoenix error | %s' % e) raise exception(e) return self.phoenix_cursor.fetchall() def disconnect_from_phoenix(self): """ 断开phoenix连接 """ print('executing : disconnect from hbase') self.phoenix_cursor.close() self.phoenix_conn.close() class mysolr(object): """ =================================================================== ===================== mysolr ========================= =================================================================== """ def __init__(self): self.solr_conn = none self.base_url = none def connect_to_solr(self, address, selector): """连接到solr服务器. """ self.base_url = 'http://{0}/solr/{1}/'.format(address, selector) self.solr_conn = pysolr.solr(self.base_url) print('executing : connect to solr | %s' % self.base_url) def get_solr_data(self, query): """ 获取solr数据 """ results = list() print('executing : search | %s' % query) try: items = self.solr_conn.search(query) for item in items: results.append(item) except exception as e: print('solr error | %s' % e) raise exception(e) return results def add_solr_data(self, data): """ 添加solr数据 """ print('executing : add | %s' % data) try: self.solr_conn.add([data]) self.solr_conn.commit() except exception as e: print('solr error | %s' % e) raise exception(e) def del_solr_byid(self, data): """ 删除solr数据 """ print('executing : del | %s' % data) try: self.solr_conn.delete(id=data) self.solr_conn.commit() except exception as e: print('solr error | %s' % e) raise exception(e) if __name__ == '__main__': print('this is test.') ms = mysolr() me = myels() mp = myphoenix()
以上就是python 调用 es、solr、phoenix的示例代码的详细内容,更多关于python 调用 es、solr、phoenix的资料请关注其它相关文章!