Kafka 表引擎简介

  • 对接 Kafka 系统，订阅 Kafka 主题，接收消息

创建 Kafka 引擎表

  • 声明

     1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

    ENGINE = Kafka()
SETTINGS
    kafka_broker_list = 'host:port, ...',
    kafka_topic_list = 'topic1, topic2, ...',
    kafka_group_name = 'consumer_group_name',
    kafka_format = 'data_format',
    [kafka_row_delimiter = 'delimiter_symbol',]
    [kafka_schema = '',]
    [kafka_num_consumers = N,]
    [kafka_skip_broken_messages = N,]
    [kafka_commit_every_batch =N];

  • kafka_broker_list: kafka 节点地址列表，用逗号分隔

  • kafka_topic_list: 订阅的主题列表，用逗号分隔

  • kafka_group_name: 消费组名称，引擎会依据此名称创建消费组

  • kafka_format: 消息格式，如 TSV、JSONEachRow、CSV 等

  • kafka_row_delimiter: 一行数据的结束符，默认 ‘\0’

  • kafka_schema: kafka schema 参数

  • kafka_num_consumers: 消费者数量，默认 1

  • kafka_skip_broken_messages: 允许跳过的错误消息数量，默认0

  • kafka_commit_every_batch: kafka commit 频率，默认 0，即整个 Block 完全写入后才 commit

Kafka 表引擎其他参数

  • stream_poll_timeout_ms: 默认每 500ms 消费一次数据，写入缓存

  • 刷新缓存触发条件:

    • 一个数据块(kafka_max_block_size，默认 65536)写入完成
    • 等待 7500 毫秒(stream_flush_interval_ms)

  • config.xml 中的 librdkafka 配置，参考 https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md

    1
2
3

    <kafka>
    <auto_offset_reset>smallest</auto_offset_reset>
</kafka>

Kafka 引擎表一般用法

  • 创建 Kafka 引擎表，充当数据管道

  • 创建 MergeTree 引擎表，用于查询

  • 创建物化视图，同步 kafka 数据到 MergeTree 引擎表

    1
2

    CREATE MATERIALIZED VIEW kafka_view TO mergetree_table
    AS SELECT col1, col2, ... FROM kafka_table;

  • 要停止数据同步，可以删除视图，也可以卸载视图

    1
2
3
4

    -- 删除
DROP TABLE kafka_view;
-- 卸载
DETACH TABLE kafka_view;

  • 恢复数据同步，装载视图

    1
2

    ATTACH MATERIALIZED VIEW kafka_view TO mergetree_table
    AS SELECT col1, col2, ... FROM kafka_table;