python pandas.DataFrame 直接写入Clickhouse

import pandas as pd
import sqlalchemy
from clickhouse_sqlalchemy import Table, engines
from sqlalchemy import create_engine, MetaData, Column
import urllib.parsehost = '1.1.1.1'
user = 'default'
password = 'default'
db = 'test'
port = 8123 # http连接端口
engine = create_engine('clickhouse://{user}:{password}@{host}:{port}/{db}'.format(user = user,host = host,password = urllib.parse.quote_plus(password),db = db,port = port),pool_size = 30,max_overflow = 0,pool_pre_ping=True , pool_recycle= 3600)
port = 9000 # Tcp/Ip连接端口
engine1 = create_engine('clickhouse+native://{user}:{password}@{host}:{port}/{db}'.format(user = user,host = host,password = urllib.parse.quote_plus(password),db = db,port = port),pool_size = 30,max_overflow = 0,pool_pre_ping=True , pool_recycle=3600)# https://github.com/xzkostyan/clickhouse-sqlalchemy/issues/129
# 参考文档https://github.com/xzkostyan/clickhouse-sqlalchemy
# pip install sqlalchemy -i https://pypi.tuna.tsinghua.edu.cn/simple
# pip install clickhouse-sqlalchemy -i https://pypi.tuna.tsinghua.edu.cn/simpleclass ClickhouseDf(object):def __init__(self, **kwargs):self.engines_dict = {"MergeTree": engines.MergeTree,"AggregatingMergeTree": engines.AggregatingMergeTree,"GraphiteMergeTree": engines.GraphiteMergeTree,"CollapsingMergeTree": engines.CollapsingMergeTree,"VersionedCollapsingMergeTree": engines.VersionedCollapsingMergeTree,"SummingMergeTree": engines.SummingMergeTree,"ReplacingMergeTree": engines.ReplacingMergeTree,"Distributed": engines.Distributed,"ReplicatedMergeTree": engines.ReplicatedMergeTree,"ReplicatedAggregatingMergeTree": engines.ReplicatedAggregatingMergeTree,"ReplicatedCollapsingMergeTree": engines.ReplicatedCollapsingMergeTree,"ReplicatedVersionedCollapsingMergeTree": engines.ReplicatedVersionedCollapsingMergeTree,"ReplicatedReplacingMergeTree": engines.ReplicatedReplacingMergeTree,"ReplicatedSummingMergeTree": engines.ReplicatedSummingMergeTree,"View": engines.View,"MaterializedView": engines.MaterializedView,"Buffer": engines.Buffer,"TinyLog": engines.TinyLog,"Log": engines.Log,"Memory": engines.Memory,"Null": engines.Null,"File": engines.File}self.table_engine = kwargs.get("table_engine", "MergeTree")  # 默认引擎选择if self.table_engine not in self.engines_dict.keys():raise ValueError("No engine for this table")def _createORMTable(self, df, name, con, schema, **kwargs):col_dtype_dict = {"object": sqlalchemy.Text,"int64": sqlalchemy.Integer,"int32": sqlalchemy.Integer,"int16": sqlalchemy.Integer,"int8": sqlalchemy.Integer,"int": sqlalchemy.Integer,"float64": sqlalchemy.Float,"float32": sqlalchemy.Float,"float16": sqlalchemy.Float,"float8": sqlalchemy.Float,"float": sqlalchemy.Float,}primary_key = kwargs.get("primary_key", [])df_col = df.columns.tolist()metadata = MetaData(bind=con, schema=schema)_table_check_col = []for col in df_col:col_dtype = str(df.dtypes[col])if col_dtype not in col_dtype_dict.keys():if col in primary_key:_table_check_col.append(Column(col, col_dtype_dict["object"], primary_key=True))else:_table_check_col.append(Column(col, col_dtype_dict["object"]))else:if col in primary_key:_table_check_col.append(Column(col, col_dtype_dict[col_dtype], primary_key=True))else:_table_check_col.append(Column(col, col_dtype_dict[col_dtype]))_table_check = Table(name, metadata,*_table_check_col,self.engines_dict[self.table_engine](primary_key=primary_key))return _table_checkdef _checkTable(self, name, con, schema):sql_str = f"EXISTS {schema}.{name}"if con.execute(sql_str).fetchall() == [(0,)]:return 0else:return 1def to_sql(self, df, name: str, con, schema=None, if_exists="fail",**kwargs):'''将DataFrame格式数据插入Clickhouse中{'fail', 'replace', 'append'}, default 'fail''''if self.table_engine in ["MergeTree"]:  # 表格必须有主键的引擎列表-暂时只用这种，其他未测试self.primary_key = kwargs.get("primary_key", [df.columns.tolist()[0]])else:self.primary_key = kwargs.get("primary_key", [])orm_table = self._createORMTable(df, name, con, schema, primary_key=self.primary_key)tanle_exeit = self._checkTable(name, con, schema)# 创建表if if_exists == "fail":if tanle_exeit:raise ValueError(f"table already exists :{name} ")else:orm_table.create()if if_exists == "replace":if tanle_exeit:orm_table.drop()orm_table.create()else:orm_table.create()if if_exists == "append":if not tanle_exeit:orm_table.create()# http连接下会自动将None填充为空字符串以入库，tcp/ip模式下则不会，会导致引擎报错，需要手动填充。df_dict = df.to_dict(orient="records")con.execute(orm_table.insert(), df_dict)# df.to_sql(name, con, schema, index=False, if_exists="append")if __name__ == '__main__':# 使用方法cdf = ClickhouseDf()df = pd.DataFrame({'column1': [1, 2, 3],'column2': ['A', 'B', 'C']})db = 'default'password = ''user = 'default'port = 9090host = '192.168.76.136'engine = create_engine('clickhouse+native://{user}:{password}@{host}:{port}/{db}'.format(user=user,host=host,password=urllib.parse.quote_plus(password),db=db,port=port),pool_size=30, max_overflow=0,pool_pre_ping=True, pool_recycle=3600)with engine.connect() as conn:cdf.to_sql(df, "table_name", conn, schema='default', if_exists="replace")list = engine.connect().execute("SELECT * FROM table_name").fetchall()print(list)

1) 运行需要安装包

# pip install sqlalchemy -i https://pypi.tuna.tsinghua.edu.cn/simple
# pip install clickhouse-sqlalchemy -i https://pypi.tuna.tsinghua.edu.cn/simple
 

2）cdf.to_sql(df, "table_name", conn, schema='default', if_exists="replace")

这里的 schema 一定要写，判断表是否存在 是用 

if con.execute('EXISTS default.table_name') == [(0,)]: 来判断表是否存在的

参考链接: SQLAlchemy_clickhouse_sqlalchemy-CSDN博客

https://github.com/xzkostyan/clickhouse-sqlalchemy