python如何将一个dataframe快速写入clickhouse

前言

dataframe是用python做数据分析最场景的数据结构了，如何将dataframe数据快速写入到clickhouse数据库呢？这里介绍几种方法，各有优劣势，可以结合自己的使用场景挑用。

思路与核心代码

假设df是一个dataframe数据结构，一共有5个列。

• to_records一条一条插入

df = pd.Dataframe()

for x in df.to_records(index = False): 
    sql = "INSERT INTO database.table_name(col1,col2,col3,col4,col5)VALUES"+str(tuple(x))
    # print(sql)
    client.execute(sql)
1
2
3
4
5
6

这种方法将dataframe里面每一个行看成一个记录，一条记录一条记录插入clickhouse数据表，不容易出岔子，但是一旦数据量很大，就会要很长时间；

• json化处理

df = pd.Dataframe()
...
log_json = df.to_json(orient="records", force_ascii=False) #一行一个json的数组
sql_log_insert = "INSERT INTO database.my_table FORMAT JSONEachRow " + log_json
client.execute(sql_log_insert)
1
2
3
4
5

• 整个dataframe插入

df = pd.Dataframe()

insert_query = 'INSERT INTO  database.table_name(col1,col2,col3,col4,col5)VALUES'  
values_query = ','.join([tuple(df.iloc[i]) for i in range(len(df))])  
insert_query += f' ({",".join([f"({col}, {val})" for col, val in zip(df.columns, values_query)])})'  
cursor = conn.cursor()  
cursor.execute(insert_query)
1
2
3
4
5
6
7

这种方法先利用values_query变量将每一行的数据转换为字符串形式，然后，使用INSERT语句将通过将整个DataFrame一次性插入ClickHouse中，避免了频繁的网络通信，从而提高了性能。但是需要注意的是，这种方法只适用于小型数据集。其实，也可以仿MySQL那样用to_sql的方法一次性塞入表里面

df = pd.Dataframe()
...
df.to_sql('database.table_name', conn, if_exists='append', index=False) 
1
2
3

• parallel插入

insert_query = 'INSERT INTO my_table (name, age, salary) VALUES'  
values_query = ','.join([tuple(df.iloc[i]) for i in range(len(df))])  
insert_query += f' ({",".join([f"({col}, {val})" for col, val in zip(df.columns, values_query)])})'  
cursor = conn.cursor()  
cursor.execute(insert_query, parameters=None, execution_profile='parallel')
1
2
3
4
5

对于超大型数据集，建议使用ClickHouse的并行插入功能或其他分布式技术。

优缺点分析

1，如果提前构造好了dataframe，且dataframe数据量适中，那么可以考虑一次性塞入，如果dataframe数据量足够大超出了内存容量，那么就要考虑了；
2，对于哪些容易丢失的还是一条一条插入数据库为好；