Python使用ClickHouse的實踐與踩坑記錄
ClickHouse是近年來備受關注的開源列式數(shù)據庫(DBMS),主要用于數(shù)據聯(lián)機分析(OLAP)領域,于2016年開源。目前國內社區(qū)火熱,各個大廠紛紛跟進大規(guī)模使用。
- 今日頭條,內部用ClickHouse來做用戶行為分析,內部一共幾千個ClickHouse節(jié)點,單集群最大1200節(jié)點,總數(shù)據量幾十PB,日增原始數(shù)據300TB左右。
- 騰訊內部用ClickHouse做游戲數(shù)據分析,并且為之建立了一整套監(jiān)控運維體系。
- 攜程內部從2018年7月份開始接入試用,目前80%的業(yè)務都跑在ClickHouse上。每天數(shù)據增量十多億,近百萬次查詢請求。
- 快手內部也在使用ClickHouse,存儲總量大約10PB, 每天新增200TB, 90%查詢小于3S。
在國外,Yandex內部有數(shù)百節(jié)點用于做用戶點擊行為分析,CloudFlare、Spotify等頭部公司也在使用。
ClickHouse最初是為 YandexMetrica 世界第二大Web分析平臺 而開發(fā)的。多年來一直作為該系統(tǒng)的核心組件被該系統(tǒng)持續(xù)使用著。
1. 關于ClickHouse使用實踐
首先,我們回顧一些基礎概念:
OLTP:是傳統(tǒng)的關系型數(shù)據庫,主要操作增刪改查,強調事務一致性,比如銀行系統(tǒng)、電商系統(tǒng)。OLAP:是倉庫型數(shù)據庫,主要是讀取數(shù)據,做復雜數(shù)據分析,側重技術決策支持,提供直觀簡單的結果。
1.1. ClickHouse 應用于數(shù)據倉庫場景
ClickHouse做為列式數(shù)據庫,列式數(shù)據庫更適合OLAP場景,OLAP場景的關鍵特征:
- 絕大多數(shù)是讀請求
- 數(shù)據以相當大的批次(> 1000行)更新,而不是單行更新;或者根本沒有更新。
- 已添加到數(shù)據庫的數(shù)據不能修改。
- 對于讀取,從數(shù)據庫中提取相當多的行,但只提取列的一小部分。
- 寬表,即每個表包含著大量的列
- 查詢相對較少(通常每臺服務器每秒查詢數(shù)百次或更少)
- 對于簡單查詢,允許延遲大約50毫秒
- 列中的數(shù)據相對較小:數(shù)字和短字符串(例如,每個URL 60個字節(jié))
- 處理單個查詢時需要高吞吐量(每臺服務器每秒可達數(shù)十億行)
- 事務不是必須的
- 對數(shù)據一致性要求低
- 每個查詢有一個大表。除了他以外,其他的都很小。
- 查詢結果明顯小于源數(shù)據。換句話說,數(shù)據經過過濾或聚合,因此結果適合于單個服務器的RAM中
1.2. 客戶端工具DBeaver
Clickhouse客戶端工具為dbeaver,官網為https://dbeaver.io/。
- dbeaver是免費和開源(GPL)為開發(fā)人員和數(shù)據庫管理員通用數(shù)據庫工具。[百度百科]
- 易用性是該項目的主要目標,是經過精心設計和開發(fā)的數(shù)據庫管理工具。免費、跨平臺、基于開源框架和允許各種擴展寫作(插件)。
- 它支持任何具有一個JDBC驅動程序數(shù)據庫。
- 它可以處理任何的外部數(shù)據源。
通過操作界面菜單中“數(shù)據庫”創(chuàng)建配置新連接,如下圖所示,選擇并下載ClickHouse驅動(默認不帶驅動)。

DBeaver配置是基于Jdbc方式,一般默認URL和端口如下:
jdbc:clickhouse://192.168.17.61:8123
如下圖所示。
在是用DBeaver連接Clickhouse做查詢時,有時候會出現(xiàn)連接或查詢超時的情況,這個時候可以在連接的參數(shù)中添加設置socket_timeout參數(shù)來解決問題。
jdbc:clickhouse://{host}:{port}[/{database}]?socket_timeout=600000
1.3. 大數(shù)據應用實踐
- 環(huán)境簡要說明:
- 硬件資源有限,僅有16G內存,交易數(shù)據為億級。
本應用是某交易大數(shù)據,主要包括交易主表、相關客戶信息、物料信息、歷史價格、優(yōu)惠及積分信息等,其中主交易表為自關聯(lián)樹狀表結構。
為了分析客戶交易行為,在有限資源的條件下,按日和交易點抽取、匯集交易明細為交易記錄,如下圖所示。

其中,在ClickHouse上,交易數(shù)據結構由60個列(字段)組成,截取部分如下所示:

針對頻繁出現(xiàn)“would use 10.20 GiB , maximum: 9.31 GiB”等內存不足的情況,基于ClickHouse的SQL,編寫了提取聚合數(shù)據集SQL語句,如下所示。

大約60s返回結果,如下所示:

2. Python使用ClickHouse實踐
2.1. ClickHouse第三方Python驅動clickhouse_driver
ClickHouse沒有提供官方Python接口驅動,常用第三方驅動接口為clickhouse_driver,可以使用pip方式安裝,如下所示:
pip install clickhouse_driver
Collecting clickhouse_driver
Downloading https://files.pythonhosted.org/packages/88/59/c570218bfca84bd0ece896c0f9ac0bf1e11543f3c01d8409f5e4f801f992/clickhouse_driver-0.2.1-cp36-cp36m-win_amd64.whl (173kB)
100% |████████████████████████████████| 174kB 27kB/s
Collecting tzlocal<3.0 (from clickhouse_driver)
Downloading https://files.pythonhosted.org/packages/5d/94/d47b0fd5988e6b7059de05720a646a2930920fff247a826f61674d436ba4/tzlocal-2.1-py2.py3-none-any.whl
Requirement already satisfied: pytz in d:\python\python36\lib\site-packages (from clickhouse_driver) (2020.4)
Installing collected packages: tzlocal, clickhouse-driver
Successfully installed clickhouse-driver-0.2.1 tzlocal-2.1使用的client api不能用了,報錯如下:
File "clickhouse_driver\varint.pyx", line 62, in clickhouse_driver.varint.read_varint
File "clickhouse_driver\bufferedreader.pyx", line 55, in clickhouse_driver.bufferedreader.BufferedReader.read_one
File "clickhouse_driver\bufferedreader.pyx", line 240, in clickhouse_driver.bufferedreader.BufferedSocketReader.read_into_buffer
EOFError: Unexpected EOF while reading bytes
Python驅動使用ClickHouse端口9000。
ClickHouse服務器和客戶端之間的通信有兩種協(xié)議:http(端口8123)和本機(端口9000)。DBeaver驅動配置使用jdbc驅動方式,端口為8123。
ClickHouse接口返回數(shù)據類型為元組,也可以返回Pandas的DataFrame,本文代碼使用的為返回DataFrame。
collection = self.client.query_dataframe(self.query_sql)
2.2. 實踐程序代碼
由于我本機最初資源為8G內存(現(xiàn)擴到16G),以及實際可操作性,分批次取數(shù)據保存到多個文件中,每個文件大約為1G。
# -*- coding: utf-8 -*-
'''
Created on 2021年3月1日
@author: xiaoyw
'''
import pandas as pd
import json
import numpy as np
import datetime
from clickhouse_driver import Client
#from clickhouse_driver import connect
# 基于Clickhouse數(shù)據庫基礎數(shù)據對象類
class DB_Obj(object):
'''
192.168.17.61:9000
ebd_all_b04.card_tbl_trade_m_orc
'''
def __init__(self, db_name):
self.db_name = db_name
host='192.168.17.61' #服務器地址
port ='9000' #'8123' #端口
user='***' #用戶名
password='***' #密碼
database=db_name #數(shù)據庫
send_receive_timeout = 25 #超時時間
self.client = Client(host=host, port=port, database=database) #, send_receive_timeout=send_receive_timeout)
#self.conn = connect(host=host, port=port, database=database) #, send_receive_timeout=send_receive_timeout)
def setPriceTable(self,df):
self.pricetable = df
def get_trade(self,df_trade,filename):
print('Trade join price!')
df_trade = pd.merge(left=df_trade,right=self.pricetable[['occurday','DIM_DATE','END_DATE','V_0','V_92','V_95','ZDE_0','ZDE_92',
'ZDE_95']],how="left",on=['occurday'])
df_trade.to_csv(filename,mode='a',encoding='utf-8',index=False)
def get_datas(self,query_sql):
n = 0 # 累計處理卡客戶數(shù)據
k = 0 # 取每次DataFrame數(shù)據量
batch = 100000 #100000 # 分批次處理
i = 0 # 文件標題順序累加
flag=True # 數(shù)據處理解釋標志
filename = 'card_trade_all_{}.csv'
while flag:
self.query_sql = query_sql.format(n, n+batch)
print('query started')
collection = self.client.query_dataframe(self.query_sql)
print('return query result')
df_trade = collection #pd.DataFrame(collection)
i=i+1
k = len(df_trade)
if k > 0:
self.get_trade(df_trade, filename.format(i))
n = n + batch
if k == 0:
flag=False
print('Completed ' + str(k) + 'trade details!')
print('Usercard count ' + str(n) )
return n
# 價格變動數(shù)據集
class Price_Table(object):
def __init__(self, cityname, startdate):
self.cityname = cityname
self.startdate = startdate
self.filename = 'price20210531.csv'
def get_price(self):
df_price = pd.read_csv(self.filename)
......
self.price_table=self.price_table.append(data_dict, ignore_index=True)
print('generate price table!')
class CardTradeDB(object):
def __init__(self,db_obj):
self.db_obj = db_obj
def insertDatasByCSV(self,filename):
# 存在數(shù)據混合類型
df = pd.read_csv(filename,low_memory=False)
# 獲取交易記錄
def getTradeDatasByID(self,ID_list=None):
# 字符串過長,需要使用'''
query_sql = '''select C.carduser_id,C.org_id,C.cardasn,C.occurday as
......
limit {},{})
group by C.carduser_id,C.org_id,C.cardasn,C.occurday
order by C.carduser_id,C.occurday'''
n = self.db_obj.get_datas(query_sql)
return n
if __name__ == '__main__':
PTable = Price_Table('湖北','2015-12-01')
PTable.get_price()
db_obj = DB_Obj('ebd_all_b04')
db_obj.setPriceTable(PTable.price_table)
CTD = CardTradeDB(db_obj)
df = CTD.getTradeDatasByID()
返回本地文件為:

3. 小結一下
ClickHouse在OLAP場景下應用,查詢速度非???,需要大內存支持。Python第三方clickhouse-driver 驅動基本滿足數(shù)據處理需求,如果能返回Pandas DataFrame最好。
ClickHouse和Pandas聚合都是非??斓?,ClickHouse聚合函數(shù)也較為豐富(例如文中anyLast(x)返回最后遇到的值),如果能通過SQL聚合的,還是在ClickHouse中完成比較理想,把更小的結果集反饋給Python進行機器學習。
操作ClickHouse刪除指定數(shù)據
def info_del2(i):
client = click_client(host='地址', port=端口, user='用戶名', password='密碼',
database='數(shù)據庫')
sql_detail='alter table SS_GOODS_ORDER_ALL delete where order_id='+str(i)+';'
try:
client.execute(sql_detail)
except Exception as e:
print(e,'刪除商品數(shù)據失敗')在進行數(shù)據刪除的時候,python操作clickhou和mysql的方式不太一樣,這里不能使用以往常用的%s然后添加數(shù)據的方式,必須完整的編輯一條語句,如同上面方法所寫的一樣,傳進去的參數(shù)統(tǒng)一使用str類型
以上為個人經驗,希望能給大家一個參考,也希望大家多多支持腳本之家。
相關文章
Python辦公自動化之教你用Python批量識別發(fā)票并錄入到Excel表格中
今天來分享一篇辦公干貨文章,對于財務專業(yè)等學生或者公司財務人員來說,將報賬發(fā)票等匯總到excel簡直就是一個折磨.尤其是到年底的時候,公司的財務人員面對一大堆的發(fā)票簡直就是苦不堪言.正好我們學會了Python,我們應該將Python的優(yōu)勢發(fā)揮起來,需要的朋友可以參考下2021-06-06
Python restful框架接口開發(fā)實現(xiàn)
這篇文章主要介紹了Python restful框架接口開發(fā)實現(xiàn),文中通過示例代碼介紹的非常詳細,對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下2020-04-04

