Clickhouse 持久化 Kafka 数据_clickhouse kafka_format

将Kafka中数据导入ClickHouse的标准流程是：

• 在ClickHouse中建立Kafka Engine 外表，作为Kafka数据源的一个接口
• 在ClickHouse中创建普通表(通常是MergeTree系列）存储Kafka中的数据
• 在ClickHouse中创建Materialized View, 监听Kafka中的数据，并将数据写入ClickHouse存储表中

ClickHouse 提供了Kafka Engine 作为访问Kafka集群的一个接口。

CREATE TABLE kafka.Vehicle (
json String
) ENGINE = Kafka SETTINGS
kafka_broker_list = 'xxx.xxx.xxx.xxx:9092',
kafka_topic_list = 'Vehicle',
kafka_group_name = 'clickhouse1',
kafka_format = 'LineAsString',
kafka_num_consumers = 1;

必选参数：

• kafka_broker_list: 这里填写Kafka服务的broker列表，用逗号分隔
• kafka_topic_list: 这里填写Kafka topic,多个topic用逗号分隔
• kafka_group_name：这里填写消费者group名称
• kafka_format__:Kafka数据格式, ClickHouse支持的Format, 详见这里

创建存储Kafka数据的目标表，该表就是最终存储Kafka数据：

CREATE TABLE kafka_sync.VehicleWarningRaw
(
ts DateTime,
json String
) ENGINE = MergeTree()
ORDER BY ts
TTL ts + INTERVAL 1 MONTH DELETE;

创建Metrialized View 抓取数据：

CREATE MATERIALIZED VIEW kafka.Vehicle_consumer TO kafka_sync.Vehicle AS
SELECT now() AS ts, json AS json
FROM kafka.Vehicle;

也可以通过DETACH/ATTACH 来取消关联，或者重新关联到某个目标表。

DETACH TABLE kafka.Vehicle_consumer;
ATTACH TABLE kafka.Vehicle_consumer;