Golang高效批量插入数据到ClickHouse插件库clickhouse-bulk的使用

简介

ClickHouse-Bulk是一个简单的Yandex ClickHouse插入收集器，它收集请求并发送到ClickHouse服务器。主要功能包括：

• 将多个请求分组并发送到任意ClickHouse服务器
• 按时间间隔发送收集的数据
• 支持VALUES、TabSeparated等格式
• 支持多服务器发送
• 支持查询参数和请求体中的查询
• 支持用户名、密码、数据库等查询参数
• 支持基本认证

安装

二进制安装

下载适合您平台的二进制文件

Docker安装

使用docker镜像

源码安装(Go 1.23+):

git clone https://github.com/nikepan/clickhouse-bulk
cd clickhouse-bulk
go build

配置示例

{
  "listen": ":8124",
  "flush_count": 10000, // 按\n字符检查
  "flush_interval": 1000, // 毫秒
  "clean_interval": 0, // 清理内部表的频率(毫秒)
  "remove_query_id": true, // 某些驱动发送query_id会阻止批量插入
  "dump_check_interval": 300, // 尝试发送转储的间隔(秒); -1表示禁用
  "debug": false, // 记录传入请求
  "dump_dir": "dumps", // 转储未发送数据的目录(如果clickhouse出错)
  "clickhouse": {
    "down_timeout": 60, // 服务器宕机等待时间(秒)
    "connect_timeout": 10, // 服务器连接超时(秒)
    "tls_server_name": "", // 覆盖TLS证书验证的serverName
    "insecure_tls_skip_verify": false, // 不安全-完全跳过证书验证
    "servers": [
      "http://127.0.0.1:8123"
    ]
  },
  "metrics_prefix": "prefix"
}

Golang使用示例

package main

import (
	"bytes"
	"fmt"
	"net/http"
	"time"
)

func main() {
	// clickhouse-bulk服务地址
	bulkServer := "http://localhost:8124"
	
	// 示例数据
	data1 := "INSERT INTO table3 (c1, c2, c3) VALUES ('v1', 'v2', 'v3')"
	data2 := "INSERT INTO table3 (c1, c2, c3) VALUES ('v4', 'v5', 'v6')"
	
	// 创建HTTP客户端
	client := &http.Client{
		Timeout: 10 * time.Second,
	}
	
	// 发送第一个插入请求
	resp1, err := client.Post(bulkServer, "text/plain", bytes.NewBufferString(data1))
	if err != nil {
		fmt.Printf("Error sending first request: %v\n", err)
		return
	}
	defer resp1.Body.Close()
	fmt.Println("First insert sent, status:", resp1.Status)
	
	// 发送第二个插入请求
	resp2, err := client.Post(bulkServer, "text/plain", bytes.NewBufferString(data2))
	if err != nil {
		fmt.Printf("Error sending second request: %v\n", err)
		return
	}
	defer resp2.Body.Close()
	fmt.Println("Second insert sent, status:", resp2.Status)
	
	// 注意：实际clickhouse-bulk会将这两个插入合并为：
	// INSERT INTO table3 (c1, c2, c3) VALUES ('v1', 'v2', 'v3')('v4', 'v5', 'v6')
}

批量插入优化示例

package main

import (
	"bytes"
	"fmt"
	"net/http"
	"time"
)

func main() {
	bulkServer := "http://localhost:8124"
	
	// 创建批量数据
	var batch bytes.Buffer
	for i := 0; i < 1000; i++ {
		batch.WriteString(fmt.Sprintf("INSERT INTO metrics (time, value, tag) VALUES (now(), %f, 'tag%d')\n", float64(i), i%10))
	}
	
	// 发送批量请求
	client := &http.Client{Timeout: 30 * time.Second}
	resp, err := client.Post(bulkServer, "text/plain", &batch)
	if err != nil {
		fmt.Printf("Error sending batch request: %v\n", err)
		return
	}
	defer resp.Body.Close()
	
	fmt.Println("Batch insert sent, status:", resp.Status)
}

环境变量配置

可以通过环境变量配置clickhouse-bulk:

• CLICKHOUSE_BULK_DEBUG - 启用调试日志
• CLICKHOUSE_SERVERS - 逗号分隔的服务器列表
• CLICKHOUSE_FLUSH_COUNT - 插入行数阈值
• CLICKHOUSE_FLUSH_INTERVAL - 插入间隔
• CLICKHOUSE_CLEAN_INTERVAL - 内部表清理间隔
• DUMP_CHECK_INTERVAL - 重新发送转储的间隔
• CLICKHOUSE_DOWN_TIMEOUT - 服务器宕机等待时间
• CLICKHOUSE_CONNECT_TIMEOUT - 服务器连接超时
• CLICKHOUSE_TLS_SERVER_NAME - TLS证书验证的服务器名称
• CLICKHOUSE_INSECURE_TLS_SKIP_VERIFY - 跳过证书验证
• METRICS_PREFIX - prometheus指标前缀

性能提示

为了获得更好的性能，FORMAT和VALUES关键字必须大写。