python连接clickhouse两种方法

1.使用Client类

import logging
from apscheduler.schedulers.blocking import BlockingScheduler
from clickhouse_driver import Clientlogger &＃61; logging.getLogger(__name__) # 操作日志对象class CkClient():"""数据库连接的公共类&＃xff0c;初始化连接&＃xff0c;自定义查询&＃xff0c;删除等操作"""def __init__(self):self.host &＃61; "127.0.0.1"self.port &＃61; 40009self.user &＃61; "default"self.password &＃61; ""self.db &＃61; "warehouse_main_ck"self.client &＃61; Client(user&＃61;self.user, password&＃61;self.password, host&＃61;self.host, port&＃61;self.port, database&＃61;self.db)self.resultlist &＃61; []# 多条sql用;连接&＃xff0c;拆分后&＃xff0c;逐一执行def spliteSql(self, sql):sqllist &＃61; sql.split(&＃39;;&＃39;)return sqllist[0:-1]def executeSql(self, sql: str) -> list:"""支持执行多条sql"""sqllist &＃61; self.spliteSql(sql) logger.info(f"开始执行sql语句")for i in sqllist:self.resultlist &＃61; self.client.execute(i)return self.resultlistdef dojob():ck &＃61; CkClient()sql &＃61; """alter table dim_customer_statictis delete where 1&＃61;1;insert into dim_customer_statictiswith (select sum(buyer_cnt)from (select count(distinct buyer_id) as buyer_cntfrom dws_buyer_tagwhere province !&＃61; &＃39;&＃39;group by province)
) as total_cnt,(select sum(order_total_amt)from dws_buyer_tagwhere province !&＃61; &＃39;&＃39;) as total_amt
select province,buyer_cnt,buyer_percent,order_per_price,sales_percent
from (select province,total_cnt,total_amt,count(distinct buyer_id) as buyer_cnt,round(buyer_cnt / total_cnt, 6) as buyer_percent,sum(order_total_amt) as order_total_amt,round(avg(order_per_price), 2) as order_per_price,round(order_total_amt / total_amt, 6) as sales_percentfrom dws_buyer_tagwhere province !&＃61; &＃39;&＃39;group by provinceorder by buyer_cnt desc);"""# sql1 &＃61; "alter table dim_customer_statictis delete where 1&＃61;1;"result &＃61; ck.executeSql(sql)# result1 &＃61; ck.executeSql(sql1)print("res:", result)def data_main():# 创建调度器&＃xff1a;BlockingSchedulerscheduler &＃61; BlockingScheduler()# 添加任务,定时启动,每天9&＃xff1a;49scheduler.add_job(dojob, &＃39;cron&＃39;, hour&＃61;9, minute&＃61;49) # day_of_week&＃61;&＃39;1-5&＃39;scheduler.start()if __name__ &＃61;&＃61; "__main__":data_main()


2.使用connect函数

import logging
import traceback
from config.ckConfig import ck_configs # 这个是自定义的模块,可以是多个clickhouse连接配置&＃xff0c;ck_configs &＃61; {data&＃61;{"host":xxx,"port":xxx,"user":xxx...},data1&＃61;{"host":xxx,"port":xxx,"user":xxx...}}from clickhouse_driver import connectclass ClickhouseManger:def __init__(self, conn_conf):"""初始化连接:param conn_conf:"""self.host &＃61; conn_conf["host"]self.port &＃61; conn_conf["port"]self.pwd &＃61; conn_conf["pwd"]self.user &＃61; conn_conf["user"]self.db &＃61; conn_conf["db"]self.conn &＃61; connect(user&＃61;self.user, password&＃61;self.pwd, host&＃61;self.host, port&＃61;self.port, database&＃61;self.db)def _get_cursor(self):"""获取游标:return:"""return self.conn.cursor()def fetchone(self, sql):"""查询单条数据:param sql: sql语句:return:"""logging.info(msg&＃61;f"------clickhouse SQL: {sql}")cursor &＃61; self._get_cursor()result &＃61; {}# noinspection PyBroadExceptiontry:cursor.execute(sql)columns_types &＃61; cursor.columns_with_typescolumns &＃61; [item[0] for item in columns_types]data &＃61; cursor.fetchone()if all([columns, data]):result &＃61; dict(zip(columns, data))except BaseException:logging.error("------clickhouse SQL ERROR:{}".format(traceback.format_exc()))finally:cursor.close()return resultdef fetchmany(self, sql):"""查询多条数据:param sql: sql语句:return:"""print("sql123", sql)logging.info(msg&＃61;f"------clickhouse SQL: {sql}")cursor &＃61; self._get_cursor()result &＃61; []# noinspection PyBroadExceptiontry:cursor.execute(sql)columns_types &＃61; cursor.columns_with_typescolumns &＃61; [item[0] for item in columns_types]data_list &＃61; cursor.fetchall()if all([columns, data_list]):for data in data_list:print("---",columns,data)result.append(dict(zip(columns, data)))except BaseException:logging.error("------clickhouse SQL ERROR:{}".format(traceback.format_exc()))finally:cursor.close()return resultdef execute(self, sql):"""执行sql语句:param sql: sql语句:return:"""logging.info(msg&＃61;f"------clickhouse SQL: {sql}")cursor &＃61; self._get_cursor()result &＃61; []try:cursor.execute(sql)except Exception as e:logging.error("------clickhouse SQL ERROR:{}".format(traceback.format_exc()))finally:cursor.close()return resultdef multiClickhouseConn(ck_configs):"""多clickhouse连接:param ck_configs: clickhouse连接配置:return:"""conn_dict &＃61; {}for name, conf_inf in ck_configs.items():conn_dict[name] &＃61; ClickhouseManger(conf_inf)print("Connection (clickhouse: %s) : %s &＃61;&＃61;&＃61;&＃61; successful " % (name, conf_inf))logging.info("Connection (clickhouse %s) : %s &＃61;&＃61;&＃61;&＃61; successful " % (name, conf_inf))return conn_dict
if __name__ &＃61;&＃61; "__main__":b &＃61; multiClickhouseConn(ck_configs&＃61;ck_configs)sql &＃61; """insert into dim_customer_statictiswith (select sum(buyer_cnt)from (select count(distinct buyer_id) as buyer_cntfrom dws_buyer_tagwhere province !&＃61; &＃39;&＃39;group by province)
) as total_cnt,(select sum(order_total_amt)from dws_buyer_tagwhere province !&＃61; &＃39;&＃39;) as total_amt
select province,buyer_cnt,buyer_percent,order_per_price,sales_percent
from (select province,total_cnt,total_amt,count(distinct buyer_id) as buyer_cnt,round(buyer_cnt / total_cnt, 6) as buyer_percent,sum(order_total_amt) as order_total_amt,round(avg(order_per_price), 2) as order_per_price,round(order_total_amt / total_amt, 6) as sales_percentfrom dws_buyer_tagwhere province !&＃61; &＃39;&＃39;group by provinceorder by buyer_cnt desc)"""print(b[&＃39;data&＃39;].execute(sql))