---

[BigDataHadoop：Hadoop&OLAP数据库管理系统.V07]                           [Deployment.OLAP数据库管理系统][|ClickHouse：ClickHouse链接kafka|]

---

---

一、ClickHouse链接Kafka

### --- ClickHouse链接Kafka：此引擎与 Apache Kafka 结合使用。

~~~     # Kafka 特性：
~~~     发布或者订阅数据流。
~~~     容错存储机制。
~~~     处理流数据。

### --- 链接语法格式

~~~     # 老版格式：
Kafka(kafka_broker_list, kafka_topic_list, kafka_group_name, kafka_format
[, kafka_row_delimiter, kafka_schema, kafka_num_consumers])

~~~     # 新版格式：

Kafka SETTINGS
kafka_broker_list = 'hadoop01:9092',
kafka_topic_list = 'topic1,topic2',
kafka_group_name = 'group1',
kafka_format = 'JSONEachRow',
kafka_row_delimiter = '\n',
kafka_schema = '',
kafka_num_consumers = 2

~~~     # 必要参数：

~~~     kafka_broker_list – 以逗号分隔的 brokers 列表 (localhost:9092)。
~~~     kafka_topic_list – topic 列表 (my_topic)。
~~~     kafka_group_name – Kafka 消费组名称 (group1)。
~~~     如果不希望消息在集群中重复，请在每个分片中使用相同的组名。
~~~     kafka_format – 消息体格式。使用与 SQL 部分的 FORMAT 函数相同表示方法，
~~~     例如 JSONEachRow。了解详细信息，请参考 Formats 部分。

~~~     # 可选参数：

~~~     kafka_row_delimiter - 每个消息体（记录）之间的分隔符。
~~~     kafka_schema – 如果解析格式需要一个 schema 时，此参数必填。
~~~     例如，普罗托船长 需要 schema 文件路径以及根对象 schema.capnp:Message 的名字。
~~~     kafka_num_consumers – 单个表的消费者数量。
~~~     默认值是：1，如果一个消费者的吞吐量不足，则指定更多的消费者。
~~~     消费者的总数不应该超过 topic 中分区的数量，因为每个分区只能分配一个消费者。

二、ClickHouse连接kafka使用示例

### --- 链接说明

~~~     消费的消息会被自动追踪，因此每个消息在不同的消费组里只会记录一次。
~~~     如果希望获得两次数据，则使用另一个组名创建副本。
~~~     消费组可以灵活配置并且在集群之间同步。
~~~     例如，如果群集中有10个主题和5个表副本，则每个副本将获得2个主题。 
~~~     如果副本数量发生变化，主题将自动在副本中重新分配。
~~~     了解更多信息请访问 http://kafka.apache.org/intro。
~~~     SELECT 查询对于读取消息并不是很有用（调试除外），因为每条消息只能被读取一次。
~~~     使用物化视图创建实时线程更实用。

~~~     # 您可以这样做：

~~~     使用引擎创建一个 Kafka 消费者并作为一条数据流。
~~~     创建一个结构表。
~~~     创建物化视图，改视图会在后台转换引擎中的数据并将其放入之前创建的表中。
~~~     当 MATERIALIZED VIEW 添加至引擎，它将会在后台收集数据。
~~~     可以持续不断地从 Kafka 收集数据并通过 SELECT 将数据转换为所需要的格式。

### --- 块大小说明

~~~     为了提高性能，接受的消息被分组为 max_insert_block_size 大小的块。
~~~     如果未在 stream_flush_interval_ms 毫秒内形成块，
~~~     则不关心块的完整性，都会将数据刷新到表中。

### --- 停止接收主题数据或更改转换逻辑，请 detach 物化视图：

~~~     如果使用 ALTER 更改目标表，为了避免目标表与视图中的数据之间存在差异，推荐停止物化视图。
~~~     配置与 GraphiteMergeTree 类似，Kafka 引擎支持使用ClickHouse配置文件进行扩展配置。
~~~     可以使用两个配置键：全局(kafka) 和 主题级别 (kafka_*)。
~~~     首先应用全局配置，然后应用主题级配置（如果存在）。
                            
~~~     有关详细配置选项列表，请参阅 librdkafka配置参考。
~~~     在 ClickHouse 配置中使用下划线 (_) ，并不是使用点 (.)。例如，check.crcs=true 将是 true。
DETACH TABLE consumer;
ATTACH TABLE consumer;

    <!-- Global configuration options for all tables of Kafka engine type -->
    <kafka>
        <debug>cgrp</debug>
        <auto_offset_reset>smallest</auto_offset_reset>
    </kafka>
    
    <!-- Configuration specific for topic "logs" -->
    <kafka_logs>
        <retry_backoff_ms>250</retry_backoff_ms>
        <fetch_min_bytes>100000</fetch_min_bytes>
    </kafka_logs>

三、ClickHouse链接Kafka示例

### --- 启动相关服务

~~~     # 启动zookeeper
[root@hadoop01 ~]# ./zk.sh start
~~~     # 启动ClickHouse
[root@hadoop01 ~]# systemctl start clickhouse-server
~~~     # 启动kafka
[root@hadoop01 ~]# cd /opt/yanqi/servers/kafka_2.12/bin/
[root@hadoop01 bin]# kafka-server-start.sh -daemon ../config/server.properties

~~~     # kafka创建主题
[root@hadoop01 ~]# kafka-topics.sh --create --zookeeper hadoop01:2181/myKafka --replication-factor 1 --partitions 1 --topic ch2
Created topic "ch2".

~~~     # kafka启动生产者
[root@hadoop01 ~]# kafka-console-producer.sh --broker-list hadoop01:9092 --topic ch2
~~~ 生产数据
>2021-11-01,level1,message    
>2021-11-01,level2,message
>2021-11-01,level3,message

### --- 创建ClickHouse链接kafka表

~~~     # 在ClickHouse创建表
[root@hadoop01 ~]#  clickhouse-client -m

hadoop01 :) CREATE TABLE queue (
            q_date String,
            level String,
            message String
            )
            ENGINE = Kafka SETTINGS kafka_broker_list = 'hadoop01:9092',
                                    kafka_topic_list = 'ch2',
                                    kafka_group_name = 'kafka_group_test',
                                    kafka_format = 'CSV',
                                    kafka_num_consumers = 1,
                                    kafka_skip_broken_messages = 10;

~~~     # 可以再ClickHouse表下查看到数据
~~~     但是查询完成之后就会被删除掉，需要创建一个物化视图，来存储到数据

hadoop01 :) select * from queue;

┌─q_date─────┬─level──┬─message─┐
│ 2021-11-01 │ level1 │ message │
│ 2021-11-01 │ level2 │ message │
│ 2021-11-01 │ level3 │ message │
└────────────┴────────┴─────────┘

### --- 创建物化视图保存数据

~~~     # 创建daily表
hadoop01 :) CREATE TABLE daily (
            day Date, 
            level String, 
            message String
            ) ENGINE = MergeTree(day, (day, level),8192);

~~~     # 创建物化视图：从原始queue表中查数据，粘贴到daily表当中，就可以在daily表中查询到数据

hadoop01 :) CREATE MATERIALIZED VIEW consumer TO daily
            AS SELECT q_date, level, message
            FROM queue;

### --- 验证物化视图存储的数据

~~~     # 在kafka生产者生产数据
[root@hadoop01 ~]# kafka-console-producer.sh --broker-list hadoop01:9092 --topic ch2
>2021-11-01,level1,message    
>2021-11-01,level2,message
>2021-11-01,level3,message

~~~     # 在daily表查询数据

hadoop01 :) select * from daily;

┌────────day─┬─level──┬─message─┐
│ 0000-00-00 │ level1 │ message │
└────────────┴────────┴─────────┘
┌────────day─┬─level──┬─message─┐
│ 0000-00-00 │ level2 │ message │
└────────────┴────────┴─────────┘
┌────────day─┬─level──┬─message─┐
│ 0000-00-00 │ level3 │ message │
└────────────┴────────┴─────────┘

---

===============================END===============================

---

Walter Savage Landor:strove with none,for none was worth my strife.Nature I loved and, next to Nature, Art:I warm'd both hands before the fire of life.It sinks, and I am ready to depart                                                                                                                                                   ——W.S.Landor

---

来自为知笔记(Wiz)