如何通过ClickHouseHook向ClickHouse插入int128类型数据？

我使用ClickHouseHook向数据库插入数据，相关代码如下：

from airflow_clickhouse_plugin.hooks.clickhouse import ClickHouseHook

ch_hook = ClickHouseHook(clickhouse_conn_id=connections_name)

def update_replacingmergetree(ch_hook, table_name: str, df: pd.DataFrame):
    values = tuple(df.to_records(index=False))
    ch_hook.execute(f'INSERT INTO do_you.{table_name} VALUES', (v for v in values))

当DataFrame不含日期类型时，函数运行正常，但传入包含日期类型的DataFrame时出现错误。DataFrame结构如下：

#   Column      Non-Null Count  Dtype         
---  ------      --------------  -----         
 0   id     18 non-null     string        
 1   s_id    18 non-null     int32         
 2   month_date  18 non-null     datetime64[ns]
 3   sum    18 non-null     int32   

数据样例：

报错信息：

TypeError: unsupported operand type(s) for &: 'str' and 'int'

需要将id转换为int128类型才能避免错误，但Pandas或Numpy不支持int128类型，请问该如何处理？

解决方案 #

方法1：利用ClickHouse自动类型转换 #

直接将字符串id传入，让ClickHouse自行完成字符串到Int128的转换，同时处理日期格式：

def update_replacingmergetree(ch_hook, table_name: str, df: pd.DataFrame):
    # 将datetime类型转为ClickHouse兼容的字符串格式
    df['month_date'] = df['month_date'].dt.strftime('%Y-%m-%d %H:%M:%S')
    # 构造带列名的INSERT语句，明确字段对应关系
    columns = ', '.join(df.columns)
    placeholders = ', '.join(['%s'] * len(df.columns))
    sql = f'INSERT INTO do_you.{table_name} ({columns}) VALUES ({placeholders})'
    # 将DataFrame转为参数列表
    values = df.values.tolist()
    ch_hook.execute(sql, values)

此方法无需在Python层面处理Int128类型，依赖ClickHouse的类型解析能力，同时避免日期类型与其他字段的类型冲突。

方法2：使用clickhouse-driver原生Int128类型 #

如果ClickHouseHook底层基于clickhouse-driver，可以直接用该驱动提供的Int128类型包装id字符串：

from clickhouse_driver import types

def update_replacingmergetree(ch_hook, table_name: str, df: pd.DataFrame):
    # 将日期转为Python原生datetime对象
    df['month_date'] = df['month_date'].dt.to_pydatetime()
    # 用clickhouse-driver的Int128类型包装id字符串
    df['id'] = df['id'].apply(lambda x: types.Int128(x))
    # 按原逻辑生成记录并插入
    values = tuple(df.to_records(index=False))
    ch_hook.execute(f'INSERT INTO do_you.{table_name} VALUES', (v for v in values))

types.Int128会被ClickHouse直接识别为对应类型，解决类型不兼容问题。

方法3：文件导入（适合大数据量） #

将DataFrame导出为CSV，通过ClickHouse的文件导入语法处理，绕过Python层面的类型限制：

import io

def update_replacingmergetree(ch_hook, table_name: str, df: pd.DataFrame):
    # 转换日期格式
    df['month_date'] = df['month_date'].dt.strftime('%Y-%m-%d %H:%M:%S')
    # 将DataFrame转为CSV字符串
    csv_buffer = io.StringIO()
    df.to_csv(csv_buffer, index=False, header=False, sep=',')
    csv_buffer.seek(0)
    # 执行CSV格式导入
    ch_hook.execute(f'INSERT INTO do_you.{table_name} FORMAT CSV', csv_buffer.read())

这种方式适合大规模数据插入，性能更优且无需处理复杂类型转换。

内容的提问来源于stack exchange，提问作者John Doe