Python 调用 ES、Solr、Phoenix

1  1 #!/usr/bin/env python
2  2 # -*- coding:utf-8 -*-
3  3 # *************************************
4  4 # @Time    : 2019/8/12
5  5 # @Author  : Zhang Fan
6  6 # @Desc    : Library
7  7 # @File    : MyDatabases.py
8  8 # @Update  : 2019/8/23
9  9 # *************************************
10 10 import elasticsearch
11 11 import phoenixdb
12 12 import pysolr
13 13 import pymysql
14 14 
15 15 
16 16 class MyELS(object):
17 17     """
18 18     ===================================================================
19 19     =====================       MyELS        =========================
20 20     ===================================================================
21 21     """
22 22     def __init__(self):
23 23         self.els_conn = None
24 24 
25 25     def connect_to_els(self, host, port):
26 26         """
27 27         连接到ElasticSearch服务器.
28 28         """
29 29         self.els_conn = elasticsearch.Elasticsearch([{'host': host, 'port': port}])
30 30         print('Executing : Connect To Elastic Search | %s' % self.els_conn)
31 31 
32 32     def get_els_data(self, query, index):
33 33         """
34 34         获取ElasticSearch数据
35 35         """
36 36         print('Executing : Search  | %s' % query)
37 37         try:
38 38             rst = self.els_conn.search(index=index, q=query)
39 39             return rst['hits']
40 40         except Exception as e:
41 41             print('Elastic Search Error | %s' % e)
42 42             raise Exception(e)
43 43 
44 44 
45 45 class MyPhoenix(object):
46 46     """
47 47     ===================================================================
48 48     =====================       MyPhoenix        ======================
49 49     ===================================================================
50 50     """
51 51     def __init__(self):
52 52         self.phoenix_conn = None
53 53         self.phoenix_cursor = None
54 54 
55 55     def connect_to_phoenix(self, host, port=8765):
56 56         """
57 57         连接到phoenix服务器
58 58         """
59 59         address = 'http://{0}:{1}/'.format(host, port)
60 60         print('Executing : Connect To Phoenix | %s' % address)
61 61         self.phoenix_conn = phoenixdb.connect(address, autocommit=True)
62 62         self.phoenix_cursor = self.phoenix_conn.cursor()
63 63 
64 64     def set_schema(self, sql, schema):
65 65         """
66 66         设置schema
67 67         """
68 68         pre_sub, sub, fol_sub = sql.upper().partition('FROM')
69 69         fol_sub = ' ' + schema + '.' + fol_sub.strip()
70 70         new_sql = ''.join([pre_sub, sub, fol_sub])
71 71         return new_sql
72 72 
73 73     def execute_phoenix_sql(self, sql):
74 74         """
75 75         执行sql语句
76 76         """
77 77         # sql = self.set_schema(sql, schema)
78 78         print('Executing : Execute  | %s' % sql)
79 79         self.phoenix_cursor.execute(sql)
80 80 
81 81     def get_from_phoenix(self, sql):
82 82         """
83 83         获取phoenix数据
84 84         """
85 85         # sql = self.set_schema(sql, schema)
86 86         print('Executing : Query  | %s' % sql)
87 87         try:
88 88             self.phoenix_cursor.execute(sql)
89 89         except Exception as e:
90 90             print('Phoenix Error | %s' % e)
91 91             raise Exception(e)
92 92         return self.phoenix_cursor.fetchall()
93 93 
94 94     def disconnect_from_phoenix(self):
95 95         """
96 96         断开phoenix连接
97 97         """
98 98         print('Executing : Disconnect From HBase')
99 99         self.phoenix_cursor.close()
100100         self.phoenix_conn.close()
101101 
102102 
103103 class MySolr(object):
104104     """
105105     ===================================================================
106106     =====================       MySolr        =========================
107107     ===================================================================
108108     """
109109     def __init__(self):
110110         self.solr_conn = None
111111         self.base_url = None
112112 
113113     def connect_to_solr(self, address, selector):
114114         """连接到solr服务器.
115115         """
116116         self.base_url = 'http://{0}/solr/{1}/'.format(address, selector)
117117         self.solr_conn = pysolr.Solr(self.base_url)
118118         print('Executing : Connect To Solr | %s' % self.base_url)
119119 
120120     def get_solr_data(self, query):
121121         """
122122         获取solr数据
123123         """
124124         results = list()
125125         print('Executing : Search  | %s' % query)
126126         try:
127127             items = self.solr_conn.search(query)
128128             for item in items:
129129                 results.append(item)
130130         except Exception as e:
131131             print('Solr Error | %s' % e)
132132             raise Exception(e)
133133         return results
134134 
135135     def add_solr_data(self, data):
136136         """
137137         添加solr数据
138138         """
139139         print('Executing : add  | %s' % data)
140140         try:
141141             self.solr_conn.add([data])
142142             self.solr_conn.commit()
143143         except Exception as e:
144144             print('Solr Error | %s' % e)
145145             raise Exception(e)
146146 
147147     def del_solr_byId(self, data):
148148         """
149149         删除solr数据
150150         """
151151         print('Executing : del  | %s' % data)
152152         try:
153153             self.solr_conn.delete(id=data)
154154             self.solr_conn.commit()
155155         except Exception as e:
156156             print('Solr Error | %s' % e)
157157             raise Exception(e)
158158 
159159 
160160 if __name__ == '__main__':
161161     print('This is test.')
162162     ms = MySolr()
163163     me = MyELS()
164164     mp = MyPhoenix()