Python에서 ClickHouse를 사용하는 방법-파이썬 튜토리얼-php.cn

클릭하우스는 최근 많은 관심을 받고 있는 오픈소스 컬럼형 데이터베이스(DBMS)로 주로 데이터 온라인 분석(OLAP) 분야에서 활용되며 2016년 오픈소스로 공개됐다. 현재 국내 커뮤니티는 호황을 누리고 있으며 주요 제조업체가 이를 대규모로 추적하여 사용하고 있습니다.

Toutiao는 사용자 행동 분석을 위해 내부적으로 ClickHouse를 사용합니다. 내부적으로는 총 수천 개의 ClickHouse 노드가 있으며, 단일 클러스터에는 최대 1,200개의 노드가 있으며, 총 데이터 양은 수십 PB이며 일일 증가량은 증가합니다. 데이터 용량은 약 300TB입니다.
Tencent는 게임 데이터 분석을 위해 내부적으로 ClickHouse를 사용하고 있으며 이에 대한 완벽한 모니터링 및 운영 시스템을 구축했습니다.
씨트립은 2018년 7월 시범운영을 시작으로 내부 업무의 80%를 ClickHouse 데이터베이스로 이전했습니다. 데이터는 매일 10억 이상 증가하고 쿼리 요청은 거의 100만 건에 달합니다.
Kuaishou도 내부적으로 ClickHouse를 사용하고 있으며 총 저장 용량은 약 10PB이며 매일 200TB가 추가되며 쿼리의 90%가 3S 미만입니다.

해외에서 Yandex는 사용자 클릭 행동을 분석하는 데 사용되는 수백 개의 노드를 보유하고 있으며 CloudFlare 및 Spotify와 같은 선도적인 회사에서도 이를 사용하고 있습니다.

ClickHouse는 원래 세계에서 두 번째로 큰 웹 분석 플랫폼인 YandexMetrica를 개발하기 위해 개발되었습니다. 수년간 시스템의 핵심 구성요소로 지속적으로 사용되어 왔습니다.

1. ClickHouse의 사용 관행에 대해

먼저 몇 가지 기본 개념을 검토해 보겠습니다.

OLTP: 추가, 삭제, 수정 및 쿼리를 주로 수행하는 전통적인 관계형 데이터베이스입니다. , 은행 시스템, 전자상거래 시스템 등 거래 일관성을 강조합니다. OLTP：是传统的关系型数据库，主要操作增删改查，强调事务一致性，比如银行系统、电商系统。
OLAP

OLAP: 주로 데이터를 읽고 복잡한 데이터 분석을 수행하며 기술적 의사결정 지원에 중점을 두고 직관적이고 간단한 결과를 제공하는 웨어하우스형 데이터베이스입니다.

1.1 ClickHouse는 데이터 웨어하우스 시나리오에 적용됩니다

ClickHouse는 OLAP 시나리오에 더 적합한 열형 데이터베이스입니다.
대부분이 읽기 요청입니다
데이터는 단일 행이 아닌 상당히 큰 배치(행 1000개 이상)로 업데이트되거나 전혀 업데이트되지 않습니다.
데이터베이스에 추가된 데이터는 수정할 수 없습니다.
읽기의 경우 데이터베이스에서 꽤 많은 행을 가져오지만 열의 일부만 가져옵니다.
넓은 테이블, 즉 각 테이블에 많은 수의 열이 포함됨
쿼리가 상대적으로 적음(일반적으로 서버당 초당 수백 개 이하)
간단한 쿼리의 경우 약 50밀리초의 지연을 허용합니다.
열의 데이터는 비교적 작습니다: 숫자 및 짧은 문자열(예: URL당 60바이트)
단일 쿼리를 처리할 때 높은 처리량에 필요(서버당 최대 1초) 수십억 행)
트랜잭션이 필요하지 않습니다
낮은 데이터 일관성 요구 사항
쿼리당 하나의 큰 테이블. 그 사람 빼고는 다 작아요.

쿼리 결과가 원본 데이터보다 훨씬 작습니다. 즉, 결과가 단일 서버의 RAM에 맞도록 데이터를 필터링하거나 집계합니다.

1.2 클라이언트 도구 DBeaver

Clickhouse 클라이언트 도구는 dbeaver이며, 공식 웹사이트는 https://dbeaver 입니다. .io/.
dbeaver는 개발자와 데이터베이스 관리자를 위한 무료 오픈 소스(GPL) 범용 데이터베이스 도구입니다. [바이두 백과사전]
이 프로젝트의 핵심 목표는 사용 편의성을 향상시키는 것이므로 데이터베이스 관리 도구를 특별히 설계하고 개발했습니다. 오픈 소스 프레임워크를 기반으로 하는 무료 크로스 플랫폼으로 다양한 확장(플러그인) 작성이 가능합니다.
JDBC 드라이버가 있는 모든 데이터베이스를 지원합니다.

모든 외부 데이터 소스를 처리할 수 있습니다.

Python에서 ClickHouse를 사용하는 방법 아래 그림과 같이 작업 인터페이스 메뉴의 "데이터베이스"를 통해 새 연결을 생성 및 구성하고 ClickHouse 드라이버를 선택하고 다운로드합니다(기본값은 드라이버 없음).

DBeaver 구성은 Jdbc 방식을 기반으로 하며 일반적으로 기본 URL과 포트는 다음과 같습니다.

jdbc:clickhouse://192.168.17.61:8123

로그인 후 복사

아래 그림과 같습니다.

Python에서 ClickHouse를 사용하는 방법 DBeaver를 사용하여 Clickhouse에 쿼리를 연결할 때 가끔 연결 또는 쿼리 시간이 초과되는 경우가 있습니다. 이때 연결 매개변수에 소켓_timeout 매개변수를 추가하고 설정하면 문제를 해결할 수 있습니다.

jdbc:clickhouse://{host}:{port}[/{database}]?socket_timeout=600000

로그인 후 복사

1.3. 빅 데이터 응용 실습
환경에 대한 간략한 설명:

하드웨어 리소스는 제한되어 있으며 메모리는 16G에 불과하며 거래 데이터는 수십억입니다.

이 애플리케이션은 특정 거래 빅데이터로, 주로 거래 마스터 테이블, 관련 고객 정보, 자재 정보, 과거 가격, 할인 및 포인트 정보 등을 포함합니다. 주요 거래 테이블은 자체 연관 트리 테이블 구조입니다. .

Python에서 ClickHouse를 사용하는 방법 고객의 거래 행태를 분석하기 위해 제한된 자원 하에서 아래 그림과 같이 거래 내역을 날짜별, 거래 시점별로 추출하여 거래 기록으로 정리합니다.

🎜🎜🎜

其中，在ClickHouse上，交易数据结构由60个列（字段）组成，截取部分如下所示：

Python에서 ClickHouse를 사용하는 방법

针对频繁出现“would use 10.20 GiB , maximum: 9.31 GiB”等内存不足的情况，基于ClickHouse的SQL，编写了提取聚合数据集SQL语句，如下所示。

Python에서 ClickHouse를 사용하는 방법

大约60s返回结果，如下所示：

Python에서 ClickHouse를 사용하는 방법

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服务器和客户端之间的通信有两种协议：http（端口8123）和本机（端口9000）。DBeaver驱动配置使用jdbc驱动方式，端口为8123。

ClickHouse接口返回数据类型为元组，也可以返回Pandas的DataFrame，本文代码使用的为返回DataFrame。

collection = self.client.query_dataframe(self.query_sql)

로그인 후 복사

2.2. 实践程序代码

由于我本机最初资源为8G内存（现扩到16G），以及实际可操作性，分批次取数据保存到多个文件中，每个文件大约为1G。

# -*- coding: utf-8 -*-
&#39;&#39;&#39;
Created on 2021年3月1日
@author: xiaoyw
&#39;&#39;&#39;
import pandas as pd
import json
import numpy as np
import datetime
from clickhouse_driver import Client
#from clickhouse_driver import connect
# 基于Clickhouse数据库基础数据对象类
class DB_Obj(object):
    &#39;&#39;&#39;
    192.168.17.61:9000
    ebd_all_b04.card_tbl_trade_m_orc
    &#39;&#39;&#39;
    def __init__(self, db_name):
        self.db_name = db_name
        host=&#39;192.168.17.61&#39; #服务器地址
        port =&#39;9000&#39; #&#39;8123&#39; #端口
        user=&#39;***&#39; #用户名
        password=&#39;***&#39; #密码
        database=db_name #数据库
        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(&#39;Trade join price!&#39;)
        df_trade = pd.merge(left=df_trade,right=self.pricetable[[&#39;occurday&#39;,&#39;DIM_DATE&#39;,&#39;END_DATE&#39;,&#39;V_0&#39;,&#39;V_92&#39;,&#39;V_95&#39;,&#39;ZDE_0&#39;,&#39;ZDE_92&#39;,
                              &#39;ZDE_95&#39;]],how="left",on=[&#39;occurday&#39;])
        df_trade.to_csv(filename,mode=&#39;a&#39;,encoding=&#39;utf-8&#39;,index=False)
    def get_datas(self,query_sql):          
        n = 0 # 累计处理卡客户数据
        k = 0 # 取每次DataFrame数据量
        batch = 100000 #100000 # 分批次处理
        i = 0 # 文件标题顺序累加
        flag=True # 数据处理解释标志
        filename = &#39;card_trade_all_{}.csv&#39;
        while flag:
            self.query_sql = query_sql.format(n, n+batch) 
            print(&#39;query started&#39;)
            collection = self.client.query_dataframe(self.query_sql)
            print(&#39;return query result&#39;)
            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(&#39;Completed &#39; + str(k) + &#39;trade details!&#39;)
            print(&#39;Usercard count &#39; + str(n) )    
               
        return n                
# 价格变动数据集
class Price_Table(object):
    def __init__(self, cityname, startdate):
        self.cityname = cityname
        self.startdate = startdate
        self.filename = &#39;price20210531.csv&#39;
        
    def get_price(self):
        df_price = pd.read_csv(self.filename)
        ......
            self.price_table=self.price_table.append(data_dict, ignore_index=True)    
            
        print(&#39;generate price table!&#39;)   
class CardTradeDB(object):
    def __init__(self,db_obj): 
        self.db_obj = db_obj
        
    def insertDatasByCSV(self,filename):
        # 存在数据混合类型
        df = pd.read_csv(filename,low_memory=False)
        
    # 获取交易记录    
    def getTradeDatasByID(self,ID_list=None):
        # 字符串过长，需要使用&#39;&#39;&#39;
        query_sql = &#39;&#39;&#39;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&#39;&#39;&#39;
        
        
        n = self.db_obj.get_datas(query_sql)
        
        return n
                    
if __name__ == &#39;__main__&#39;:
    PTable = Price_Table(&#39;湖北&#39;,&#39;2015-12-01&#39;)   
    PTable.get_price()  
    
    db_obj = DB_Obj(&#39;ebd_all_b04&#39;)
    db_obj.setPriceTable(PTable.price_table)
    CTD = CardTradeDB(db_obj)
    df = CTD.getTradeDatasByID()

로그인 후 복사

返回本地文件为：

Python에서 ClickHouse를 사용하는 방법

3. 小结一下

ClickHouse运用于OLAP场景时，拥有出色的查询速度，但需要具备大内存支持。Python第三方clickhouse-driver 驱动基本满足数据处理需求，如果能返回Pandas DataFrame最好。

ClickHouse和Pandas聚合都是非常快的，ClickHouse聚合函数也较为丰富（例如文中anyLast(x)返回最后遇到的值），如果能通过SQL聚合的，还是在ClickHouse中完成比较理想，把更小的结果集反馈给Python进行机器学习。

操作ClickHouse删除指定数据

def info_del2(i):
    client = click_client(host=&#39;地址&#39;, port=端口, user=&#39;用户名&#39;, password=&#39;密码&#39;,
                          database=&#39;数据库&#39;)
    sql_detail=&#39;alter table SS_GOODS_ORDER_ALL delete where order_id=&#39;+str(i)+&#39;;&#39;
    try:
        client.execute(sql_detail)
    except Exception as e:
        print(e,&#39;删除商品数据失败&#39;)

로그인 후 복사

在进行数据删除的时候，python操作clickhou和mysql的方式不太一样，这里不能使用以往常用的%s然后添加数据的方式，必须完整的编辑一条语句，如同上面方法所写的一样，传进去的参数统一使用str类型

위 내용은 Python에서 ClickHouse를 사용하는 방법의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!