【ClickHouse】批量写入ClickHouse 的几种方式

ClickHouse没有官方的Python接口，有个第三方的库，叫clickhouse-driver，笔者所知道的将数据批量写入的方式不是很多，这里列举最常见的3种方式。第一种方式 CSV文件cat /dev/shm/data | clickhouse-client -h 123.456.1.X --query="insert into database.table FORMAT CSV"```或

幸运的小新粉

15082人浏览 · 2021-12-12 11:32:08

幸运的小新粉 · 2021-12-12 11:32:08 发布

ClickHouse没有官方的Python接口，有个第三方的库，叫clickhouse-driver，笔者所知道的将数据批量写入的方式不是很多，这里列举最常见的3种方式。

第一种方式 CSV文件

cat /dev/shm/data | clickhouse-client -h 123.456.1.X --query="insert into database.table FORMAT CSV"```
或
clickhouse-client -h 192.168.1.X --query="insert into database.table FORMAT CSV" < /dev/shm/data.csv

-h 表示的是要连接的的数据库
导入的CSV文件有字段名称时

cat /dev/shm/data.csv | clickhouse-client --query="INSERT INTO 数据库.表名 FORMAT CSVWithNames"

不过这种方式需要从磁盘指定文件写入，不是很实用，不是很推荐。

第二种方式 MySQL导入

这种方式比较常见，相当于先将数据保存在mysql中，然后将数据迁移到clickhouse。
需要注意的是建表的时候需要注意两遍的字段类型，特别是 decimal 的类型，clickhouse建表需要注意更改为 Float 类型。

def get_decimal_cols(table):
    sql = "describe {table};".format(table=table)
    df = common_utils.read_db_data(sql)
    cols = df[(df["Type"].str.startswith("decimal"))]["Field"].values.tolist()
    return cols

# 修改列类型
decimal_cols = get_decimal_cols(src_table)
if len(decimal_cols) > 0:
    alert_sql_tpl = "ALTER TABLE {tmp_table} MODIFY COLUMN {col_name} {col_type}"
    for col_name in decimal_cols:
        alter_sql = alert_sql_tpl.format(
            tmp_table=tmp_table, col_name=col_name, col_type="Float64", postfix=postfix)
        logging.info("alter sql= %s" % alter_sql)
        client.execute(alter_sql)

这篇文章写得很好，推荐一下：
使用ClickHouse表函数将MySQL数据导入到ClickHouse

第三种方式 JSON 导入

不过有些时候不想走mysql这个中间站，又不想保存在磁盘再上传，中间一步比较多余，于是翻了下还真有。

 insert into {tmp_table} 
 FORMAT JSONEachRow
 {data_json}

这种方式也是比较简单的，并且比较大的优势就是json数据可以指定传入的列，这样灵活性是比较高的。
至于会不会出现 json数据过长的情况，看过 clickhouse-driver 的源码，事实上会将查询的query 编码后通过http 传给clickhouse服务器

“post 理论上讲是没有大小限制的，HTTP协议规范也没有进行大小限制，但实际上post所能传递的数据量大小取决于服务器的设置和内存大小”

所以比较推荐的做法是分批传，亲测性能还是很高的。


    data_list = res_df.to_dict("records")
    n = len(data_list)
    logging.info(f"insert data to {tmp_table}... total: {n}.")
    for i in range(0, n, 10000):
        sql = """
                insert into {tmp_table} 
                FORMAT JSONEachRow
                {data_json}
                """.format(
            tmp_table=tmp_table,
            data_json=json.dumps(data_list[i:i+10000]))
        client.execute(sql)