Python连接ClickHouse数据库的两种方式与实操指南

Python连接ClickHouse数据库的核心方法

第1步：安装依赖库与配置连接参数

在进行数据库交互前，需先完成驱动安装与环境准备。目的：确保通信层就绪；入口：系统终端或虚拟环境命令行；动作：执行 pip install clickhouse-driver 安装官方驱动，可选安装 psutil 用于后续示例中的系统状态采集；重要参数：确认目标服务器的IP、TCP端口（默认5440或8123）、用户名与密码；完成状态：终端无报错返回且 import clickhouse_driver 可成功加载模块。

方式一：使用Client类直接执行查询与写入

第2步：实例化Client对象并执行插入操作

该方式通过直接实例化 Client 类构建连接池，适合轻量级脚本与一次性并发任务。目的：简化连接生命周期管理；入口：脚本头部引入 from clickhouse_driver import Client；动作：传入网络参数初始化客户端，调用 client.execute(sql) 发送增删改查语句；重要参数：host、port、user、password、database、send_receive_timeout；完成状态：控制台打印成功提示且目标表中生成新记录。

from clickhouse_driver import Client
from datetime import datetime
import psutil

host_name = '192.168.50.94'
client = Client(
    host=host_name,
    database='default',
    user='default',
    password='your_password',
    send_receive_timeout=20,
    port=55666
)

now = datetime.now()
time_stamp = now.strftime('%a %b %d %H:%M:%S CST %Y')
create_at = datetime.now().strftime('%Y-%m-%d %H:%M:%S')

disk_io = psutil.disk_io_counters()
net_io = psutil.net_io_counters()

chart_name = ['磁盘IO', '网络IO']
metric_name1 = ['读(数量)', '写(数量)', '读(字节)', '写(字节)', '读(时间)', '写(时间)']
metric_value1 = [disk_io.read_count, disk_io.write_count, disk_io.read_bytes, 
                 disk_io.write_bytes, disk_io.read_time, disk_io.write_time]

metric_name2 = ['发送字节数', '接收字节数', '发送包数', '接收包数']
metric_value2 = [net_io.bytes_sent, net_io.bytes_recv, net_io.packets_sent, net_io.packets_recv]

try:
    for i in chart_name:
        if i == '磁盘IO':
            for j in metric_name1:
                val = metric_value1[metric_name1.index(j)]
                sql = f"INSERT INTO clickhouse_host_metrics777(time_stamp, host_name, chart_name, metric_name, metric_value, create_at) VALUES ('{time_stamp}', '{host_name}', '{i}', '{j}', {val}, '{create_at}')"
                client.execute(sql)
        elif i == '网络IO':
            for j in metric_name2:
                val = metric_value2[metric_name2.index(j)]
                sql = f"INSERT INTO clickhouse_host_metrics777(time_stamp, host_name, chart_name, metric_name, metric_value, create_at) VALUES ('{time_stamp}', '{host_name}', '{i}', '{j}', {val}, '{create_at}')"
                client.execute(sql)
    print('成功写入数据')
except Exception as e:
    print(str(e))

方式二：使用connect函数配合游标(cursor)操作

第3步：通过connect建立连接并使用Cursor管理事务

此模式遵循Python DB-API 2.0规范，通过显式创建游标分离连接管理与SQL执行，更适合长连接或需要手动控制提交/回滚的业务流。目的：提升连接可控性与事务处理能力；入口：引入 from clickhouse_driver import connect；动作：使用DSN字符串建立连接并调用 .cursor()，后续通过 cursor.execute() 下发SQL；重要参数：拼接格式 clickhouse://user:pass@host:port/db；完成状态：执行完毕调用 cursor.close() 与 conn.close() 释放句柄，无残留占用。

from datetime import datetime
import psutil
from clickhouse_driver import connect

host_name = '192.168.50.94'
conn = connect('clickhouse://default:your_password@192.168.50.94:55666/default')
cursor = conn.cursor()

# 此处复用上方指标采集逻辑
# ...

try:
    for i in chart_name:
        # 模拟循环执行插入
        sql = "INSERT INTO clickhouse_host_metrics777(...) VALUES (...)"
        res = cursor.execute(sql)
    cursor.close()
    print('成功写入数据')
except Exception as e:
    print(str(e))
finally:
    conn.close()

数据批量导入与性能优化建议

第4步：利用INSERT语句进行跨库或批量数据导入

当需要同步历史数据或聚合结果时，推荐使用 INSERT INTO ... SELECT 或带类型校验的批量接口。目的：减少应用层循环，利用ClickHouse服务端计算优势；入口：构造查询视图或临时表；动作：调用 client.execute(sql, types_check=True) 执行写入；重要参数：types_check=True 开启强类型校验拦截非法数据；完成状态：返回影响行数且目标分区合并正常。

from clickhouse_driver import Client

client = Client(host='localhost', port=5440, user='default', database='target_db', password='your_password')

query_ck_sql = '''SELECT *
FROM database1.table1
WHERE date = today()'''

try:
    client.execute(
        f"INSERT INTO official_table_db.official_all_table_name {query_ck_sql}",
        types_check=True
    )
except Exception as e:
    print(str(e))

批量操作需注意网络带宽与单机写入峰值。若遭遇超时，可通过延长 send_receive_timeout 或拆分批次解决；对于高频采集场景，建议结合ClickHouse的Buffer引擎或异步客户端配置进一步降载。

以上就是Python连接ClickHouse数据库的详细内容，更多关于ClickHouse数据存储与Python集成的资料请关注本站其它相关文章！