ClickHouse系列--消费kafka数据

青旅半醒 2023-01-21 03:26 12阅读 0赞

### 1.使用方式 ###

主要是使用ClickHouse的表引擎。

CREATE TABLE [IF NOT EXISTS] [db.]table_name [ON CLUSTER cluster]
    (
        name1 [type1] [DEFAULT|MATERIALIZED|ALIAS expr1],
        name2 [type2] [DEFAULT|MATERIALIZED|ALIAS expr2],
        ...
    ) ENGINE = Kafka()
    SETTINGS
        kafka_broker_list = 'host:port',
        kafka_topic_list = 'topic1,topic2,...',
        kafka_group_name = 'group_name',
        kafka_format = 'data_format'[,]
        [kafka_row_delimiter = 'delimiter_symbol',]
        [kafka_schema = '',]
        [kafka_num_consumers = N,]
        [kafka_max_block_size = 0,]
        [kafka_skip_broken_messages = N,]
        [kafka_commit_every_batch = 0,]
        [kafka_thread_per_consumer = 0]

kafka\_broker\_list ：逗号分隔的brokers地址 (localhost:9092).  
kafka\_topic\_list ：Kafka 主题列表，多个主题用逗号分隔.  
kafka\_group\_name ：消费者组.  
kafka\_format – Message format. 比如JSONEachRow、JSON、CSV等等

### 2.示例 ###

2.1在kafka中创建user\_behavior主题，并向该主题写入数据，数据示例为：

{ "user_id":63401,"item_id":6244,"cat_id":143,"action":"pv","province":3,"ts":1573445919}
    { "user_id":9164,"item_id":2817,"cat_id":611,"action":"fav","province":28,"ts":1573420486}
    { "user_id":63401,"item_id":6244,"cat_id":143,"action":"pv","province":3,"ts":1573445919}

在ClickHouse中创建表，选择表引擎为Kafka()，如下:

CREATE TABLE kafka_user_behavior (
        user_id UInt64 COMMENT '用户id',
        item_id UInt64 COMMENT '商品id',
        cat_id UInt16  COMMENT '品类id',
        action String  COMMENT '行为',
        province UInt8 COMMENT '省份id',
        ts UInt64      COMMENT '时间戳'
      ) ENGINE = Kafka()
        SETTINGS
        kafka_broker_list = 'cdh04:9092',
        kafka_topic_list = 'user_behavior',
        kafka_group_name = 'group1',
        kafka_format = 'JSONEachRow'
    ;
    -- 查询
    cdh04 :) select * from kafka_user_behavior ;
    
    -- 再次查看数据，发现数据为空
    cdh04 :) select count(*) from kafka_user_behavior;
    
    SELECT count(*)
    FROM kafka_user_behavior
    
    ┌─count()─┐
    │       0 │
    └─────────┘

2.2通过物化视图将kafka数据导入ClickHouse

当我们一旦查询完毕之后，ClickHouse会删除表内的数据，其实Kafka表引擎只是一个数据管道，我们可以通过物化视图的方式访问Kafka中的数据。

*  首先创建一张Kafka表引擎的表，用于从Kafka中读取数据
 *  然后再创建一张普通表引擎的表，比如MergeTree，面向终端用户使用
 *  最后创建物化视图，用于将Kafka引擎表实时同步到终端用户所使用的表中

--  创建Kafka引擎表
     CREATE TABLE kafka_user_behavior_src (
        user_id UInt64 COMMENT '用户id',
        item_id UInt64 COMMENT '商品id',
        cat_id UInt16  COMMENT '品类id',
        action String  COMMENT '行为',
        province UInt8 COMMENT '省份id',
        ts UInt64      COMMENT '时间戳'
      ) ENGINE = Kafka()
        SETTINGS
        kafka_broker_list = 'cdh04:9092',
        kafka_topic_list = 'user_behavior',
        kafka_group_name = 'group1',
        kafka_format = 'JSONEachRow'
    ;
    
    -- 创建一张终端用户使用的表
     CREATE TABLE kafka_user_behavior (
        user_id UInt64 COMMENT '用户id',
        item_id UInt64 COMMENT '商品id',
        cat_id UInt16  COMMENT '品类id',
        action String  COMMENT '行为',
        province UInt8 COMMENT '省份id',
        ts UInt64      COMMENT '时间戳'
      ) ENGINE = MergeTree()
        ORDER BY user_id
    ;
    -- 创建物化视图，同步数据
    CREATE MATERIALIZED VIEW user_behavior_consumer TO kafka_user_behavior
        AS SELECT * FROM kafka_user_behavior_src ;
    -- 查询，多次查询，已经被查询的数据依然会被输出
    cdh04 :) select * from kafka_user_behavior;

**Note:  
Kafka消费表不能直接作为结果表使用。Kafka消费表只是用来消费Kafka数据，没有真正的存储所有数据。**

**这里还有一个疑问：  
在众多资料中，kafka示例消息都是最简单的json格式，如果消息格式是复杂类型呢？是否支持？**