Clickhouse 读取Kafka数据写入表

1、启动kafka和zookeeper

2、创建一个生产者

sh kafka2.6-start.sh producer demo

3、在ck中创建一个表

#创建一个流引擎
CREATE TABLE queue (create_time String, level UInt64, message String) ENGINE = Kafka('hadoop100:9092', 'demo', 'group1', 'CSV');

4、创建一个接受kafka引擎的表

#创建一个表结构
CREATE TABLE daily ( day Date,level UInt64, total UInt64) ENGINE = SummingMergeTree(day, (day, level), 8192);
或
CREATE TABLE daily1 ( create_time String, level UInt64, message String) ENGINE = MergeTree() PARTITION BY toYYYYMM(toDateTime(create_time)) ORDER BY create_time; 

5、创建一个物化视图，将Kafka流引擎数据导入MergeTree表中

#创建物化视图到 结构表daily中
CREATE MATERIALIZED VIEW consumer TO daily AS SELECT toDate(toDateTime(create_time)) AS day, level, count() as total FROM queue GROUP BY day, level;
或

CREATE MATERIALIZED VIEW consumer1 TO daily1 AS SELECT create_time, level, message FROM queue;

6、测试数据

"2022-04-11 17:33:21",1,"ceshi"

"2022-04-11",1,"ceshi"

7、查看表数据

select * from consumer1;

select * from daily1;