这篇说说如何从 Kafka 中实时消费数据到腾讯云数据仓库 TCHouse-C。
一、前提条件
数据源 Kafka 集群和目的端腾讯云数据仓库 TCHouse-C 集群必须在同一个 VPC 下。
二、操作步骤
1. 登录 腾讯云数据仓库 TCHouse-C 集群，创建 Kafka 消费表。
CREATE TABLE queue ( timestamp UInt64, level String, message String ) ENGINE = Kafka SETTINGS kafka_broker_list = 'localhost:9092', kafka_topic_list = 'topic', kafka_group_name = 'group', kafka_format = 'JSONEachRow', kafka_num_consumers = 1, kafka_max_block_size = 65536, kafka_skip_broken_messages = 0, kafka_auto_offset_reset = 'latest';
注意：TCHouse-C 从 21.8 版本开始不支持参数 kafka_auto_offset_reset。
常用参数说明如下：
| 名称 | 是否必选 | 说明 |
| kafka_broker_list | 是 | Kafka 服务的 broker 列表，用逗号分隔，这里建议用 Ip:port， 不要用域名（可能存在 DNS 解析问题）。 |
| kafka_topic_list | 是 | Kafka topic，多个 topic 用逗号分隔。 |
| kafka_group_name | 是 | Kafka 的消费组名称。 |
| kafka_format | 是 | Kafka 数据格式 |
| kafka_row_delimiter | 否 | 行分隔符，用于分割不同的数据行。默认为“\\n”，你也可以根据数据写入的实际分割格式进行设置。 |
| kafka_num_consumers | 否 | 单个 Kafka Engine 的消费者数量，通过增加该参数，可以提高消费数据吞吐，但总数不应超过对应 topic 的 partitions 总数。 |
| kafka_max_block_size | 否 | Kafka Engine 批量从 Kafka 读取数据的行数。 |
| kafka_skip_broken_messages | 否 | 表示忽略解析异常的 Kafka 数据的条数。如果出现了 N 条异常后，后台线程结束 默认值为0。 |
| kafka_commit_every_batch | 否 | 执行 Kafka commit 的频率，取值如下：0：完全写入一整个Block数据块的数据后才执行commit；1：每写完一个Batch批次的数据就执行一次commit。 |
| kafka_auto_offset_reset | 否 | 从哪个 offset 开始读取 Kafka 数据。取值范围：earlist，latest。 |
2. 创建 TCHouse-C 本地表（目标表）。
·如果你的集群是单副本版：
CREATE TABLE daily on cluster default_cluster( day Date, level String, total UInt64)engine = SummingMergeTree()order by int_id;
·如果你的集群是双副本版：
create table daily on cluster default_cluster( day Date, level String, total UInt64)engine = ReplicatedSummingMergeTree('/clickhouse/tables/test/test/{shard}', '{replica}')order by int_id;
·创建分布式表：
create table daily_dis on cluster default_clusterAS test.testengine = Distributed('default_cluster', 'default', 'daily', rand());
3. 创建物化视图，把 Kafka 消费表消费到的数据同步到目的表。
CREATE MATERIALIZED VIEW consumer TO dailyAS SELECT toDate(toDateTime(timestamp)) AS day, level, count() as totalFROM queue GROUP BY day, level;
4. 查询。
SELECT level, sum(total) FROM daily GROUP BY level;
三、其他
如果要停止接收主题数据或更改转换逻辑，可以进行 detach 和 attach 视图操作。
DETACH TABLE consumer; ATTACH TABLE consumer;