altinity/clickhouse-sink-connector

ClickHouse Sink Connector

概述

ClickHouse Sink Connector是一个Kafka Connect兼容的连接器，旨在实现Kafka与ClickHouse之间的高效数据集成。该连接器允许用户将Kafka主题中的流数据实时写入ClickHouse数据库，提供可靠的数据传输和灵活的配置选项，适用于构建实时数据管道和分析系统。

核心功能与特性

• 高效数据传输：优化的批处理机制，支持高吞吐量数据写入
• 灵活的数据映射：支持JSON、Avro等多种数据格式到ClickHouse表结构的映射
• 自动表创建：可配置自动创建ClickHouse表结构
• 容错机制：支持数据重试和断点续传，确保数据可靠性
• 并行处理：支持多任务并行处理，提高数据处理效率
• 自定义配置：丰富的配置选项，满足不同场景需求

使用场景

• 实时数据仓库构建：将Kafka中的流数据实时同步到ClickHouse
• 日志数据存储与分析：收集分布式系统日志并存储到ClickHouse进行分析
• 事件流处理：处理应用程序事件流并持久化到ClickHouse
• ETL数据管道：作为ETL流程的一部分，将处理后的数据加载到ClickHouse

使用方法

前提条件

• Kafka集群（0.11.0.0或更高版本）
• ClickHouse数据库（19.11或更高版本）
• Kafka Connect环境

Docker部署

基本运行命令

bash
docker run -d \
  --name=clickhouse-sink-connector \
  -e BOOTSTRAP_SERVERS=kafka:9092 \
  -e CONNECT_GROUP_ID=clickhouse-sink \
  -e CONNECT_KEY_CONVERTER=org.apache.kafka.connect.json.JsonConverter \
  -e CONNECT_VALUE_CONVERTER=org.apache.kafka.connect.json.JsonConverter \
  -e CLICKHOUSE_HOST=clickhouse \
  -e CLICKHOUSE_PORT=8123 \
  -e CLICKHOUSE_USER=default \
  -e CLICKHOUSE_PASSWORD= \
  -e CLICKHOUSE_DATABASE=default \
  altinity/clickhouse-sink-connector

Docker Compose配置

yaml
version: '3'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:latest
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  kafka:
    image: confluentinc/cp-kafka:latest
    depends_on:
      - zookeeper
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

  clickhouse:
    image: yandex/clickhouse-server:latest
    ports:
      - "8123:8123"
    volumes:
      - clickhouse_data:/var/lib/clickhouse

  clickhouse-sink:
    image: altinity/clickhouse-sink-connector
    depends_on:
      - kafka
      - clickhouse
    environment:
      BOOTSTRAP_SERVERS: kafka:9092
      CONNECT_GROUP_ID: clickhouse-sink
      CONNECT_KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter
      CONNECT_VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter
      CONNECT_KEY_CONVERTER_SCHEMAS_ENABLE: "false"
      CONNECT_VALUE_CONVERTER_SCHEMAS_ENABLE: "false"
      CONNECT_OFFSET_STORAGE_FILE_FILENAME: /tmp/connect.offsets
      CLICKHOUSE_HOST: clickhouse
      CLICKHOUSE_PORT: 8123
      CLICKHOUSE_USER: default
      CLICKHOUSE_PASSWORD: ""
      CLICKHOUSE_DATABASE: default
      TOPICS: "user_events,page_views"
      TABLES: "user_events,page_views"

volumes:
  clickhouse_data:

连接器配置

创建连接器配置文件clickhouse-sink.properties：

properties
name=clickhouse-sink-connector
connector.class=com.altinity.clickhouse.sink.connector.ClickHouseSinkConnector
tasks.max=3

# Kafka配置
topics=user_events,page_views

# ClickHouse配置
clickhouse.host=clickhouse
clickhouse.port=8123
clickhouse.user=default
clickhouse.password=
clickhouse.database=default

# 表映射
topic-to-table.map=user_events:user_events_table,page_views:page_views_table

# 数据格式配置
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=false
value.converter.schemas.enable=false

# 批处理配置
batch.size=10000
linger.ms=5000

使用Kafka Connect REST API创建连接器：

bash
curl -X POST -H "Content-Type: application/json" \
  --data @clickhouse-sink.json \
  http://connect-host:8083/connectors

配置参数

核心配置参数

高级配置参数

故障排除

1. 连接问题：检查Kafka和ClickHouse服务是否可访问，网络是否通畅
2. 认证失败：验证ClickHouse用户名和密码是否正确
3. 表不存在：启用auto.create.tables=true或手动创建所需表
4. 性能问题：调整batch.size和linger.ms参数优化性能
5. 数据格式不匹配：检查Kafka消息格式与ClickHouse表结构是否匹配

注意事项

• 在生产环境中建议设置适当的tasks.max值以充分利用并行处理能力
• 根据数据量调整批处理参数以平衡延迟和吞吐量
• 对于高可用性部署，建议配置多个连接器实例
• 定期监控连接器性能和状态，确保数据同步正常