1. 概述
在生产环境中，经常遇到将数据从消息队列Kafka写入ClickHouse集群中。本文介绍如何将Kafka中的数据导入到ClickHouse集群的方案。
Kafka 是目前应用非常广泛的开源消息中间件，一个常用的的场景就是做数据总线收集各个服务的数据，下游各种数据服务订阅消费数据，生成各种报表或数据应用等。Clickhouse 的自带了 Kafka Engine，使得 Clickhouse 和 Kafka 的集成变得非常容易。
将Kafka中数据导入ClickHouse的标准流程是：
- 在ClickHouse中建立Kafka Engine 外表，作为Kafka数据源的一个接口
- 在ClickHouse中创建普通表(通常是MergeTree系列）存储Kafka中的数据
- 在ClickHouse中创建Materialized View, 监听Kafka中的数据，并将数据写入ClickHouse存储表中；
上述三个步骤，就可以将Kafka中的数据导入到ClickHouse集群中。本文接下来，给出每个步骤的详细描述。
2. Kafka数据导入ClickHouse详细步骤
ClickHouse 提供了Kafka Engine 作为访问Kafka集群的一个接口（数据流）。有了这个接口后，导入数据就很方便了，具体步骤如下：
- 步骤1：创建Kafka Engine
CREATE TABLE source
(
`ts` DateTime,
`tag` String,
`message` String
)
ENGINE = Kafka()
SETTINGS kafka_broker_list = '172.19.0.47:9092',
kafka_topic_list = 'tag',
kafka_group_name = 'clickhouse',
kafka_format = 'JSONEachRow',
kafka_skip_broken_messages = 1,
kafka_num_consumers = 2必选参数：
kafka_broker_list: 这里填写Kafka服务的broker列表，用逗号分隔kafka_topic_list: 这里填写Kafka topic,多个topic用逗号分隔kafka_group_name：这里填写消费者group名称kafka_format____:Kafka数据格式, ClickHouse支持的Format, 详见这里 可选参数：kafka_skip_broken_messages：填写大于等于0的整数，表示忽略解析异常的Kafka数据的条数。如果出现了N条异常后，后台线程结束，Materialized View会被重新安排后台线程去监听数据kafka_num_consumers____: 单个Kafka Engine 的消费者数量，通过增加该参数，可以提高消费数据吞吐，但总数不应超过对应topic的partitions总数kafka_row_delimiter: 消息分隔符kafka_schema____:对于kafka_format需要schema定义的时候，其schema由该参数确定kafka_max_block_size: 该参数控制Kafka数据写入目标表的Block大小，超过该数值后，就将数据刷盘。- 步骤2：创建存储Kafka数据的目标表，该表就是最终存储Kafka数据 本文中，采用MergeTree来存储Kafka数据：
CREATE TABLE target
(
`ts` DateTime,
`tag` String
)
ENGINE = MergeTree()
PARTITION BY toYYYYMM(ts)
ORDER BY tag- 步骤3：创建Metrialized View 抓取数据
本文中，采用如下语句创建MV:
代码语言：txt复制CREATE MATERIALIZED VIEW source_mv TO target AS
SELECT
ts,
tag
FROM source完成上述三个步骤，我们就可以在表target中查询到来自Kafka的数据了。
3. 讨论
在上述数据导入流程中，Materialized View 起到了一个中间管道作用，将Kafka Engine代表的数据流，写入到目标表中。
实际上，一个数据流可以关联多个Materialized View, 将Kafka中的数据同时导入到多个不同目的的表中。
也可以通过DETACH/ATTACH 来取消关联，或者重新关联到某个目标表。
4. 结束
本文介绍了Kafka数据导入ClickHouse集群的步骤。想获取更多关于ClickHouse技术交流，请留言，我们将您拉入ClickHouse技术交流群。
