clickhouse-kakfa_clickhouse kafka_format

新鲜的尝试

1. 最近etl写得太多了，都是使用flink从kafka经过稍微处理写入到clickhouse
2. 看着clickhouse官网，偶然间发现了一个有趣的东西，Kafka引擎
3. clickhouse默认支持kafka表引擎，kafka一些参数：

   必要参数：
   kafka_broker_list – 以逗号分隔的 brokers 列表 (localhost:9092)。
   kafka_topic_list – topic 列表 (my_topic)。
   kafka_group_name – Kafka 消费组名称 (group1)。如果不希望消息在集群中重复，请在每个分片中使用相同的组名。
   kafka_format – 消息体格式。使用与 SQL 部分的 FORMAT 函数相同表示方法，例如 JSONEachRow。
   可选参数：
   kafka_row_delimiter - 每个消息体（记录）之间的分隔符。
   kafka_schema – 如果解析格式需要一个 schema 时，此参数必填。例如，需要 schema 文件路径以及根对象 schema.capnp:Message 的名字。
   kafka_num_consumers – 单个表的消费者数量。默认值是：1，如果一个消费者的吞吐量不足，则指定更多的消费者。消费者的总数不应该超过 topic 中分区的数量，因为每个分区只能分配一个消费者。
   

    

4. 建表示例：ck-kafka参数可在建表是指定，也可使用ck设置的默认值

     CREATE TABLE queue (
       timestamp UInt64,
       level String,
       message String
     ) ENGINE = Kafka('localhost:9092', 'topic', 'group1', 'JSONEachRow');
    
     SELECT * FROM queue LIMIT 5;
    
     CREATE TABLE queue2 (
       timestamp UInt64,
       level String,
       message String
     ) ENGINE = Kafka SETTINGS kafka_broker_list = 'localhost:9092',
                               kafka_topic_list = 'topic',
                               kafka_group_name = 'group1',
                               kafka_format = 'JSONEachRow',
                               kafka_num_consumers = 4;
    
     CREATE TABLE queue2 (
       timestamp UInt64,
       level String,
       message String
     ) ENGINE = Kafka('localhost:9092', 'topic', 'group1')
                 SETTINGS kafka_format = 'JSONEachRow',
                          kafka_num_consumers = 4;
   

    

5. 文档写得是：消费的消息会被自动追踪，并且每个消息在所有消费组里只会记录一次，保证了数据一致性语义

本地搭建环境

1. 基于docker搭建zookeeper单点
   1.  docker run --name zookeeper -p 2181:2181 -p 2888:2888 -p 3888:3888 zookeeper:3.5
   2. 命令解析自己去找吧，-p容器内部端口映射到服务器，没什么含量
2. 启动kafka，基于docker单点
   1. docker run --name kafka -d -p 9092:9092 -e KAFKA_BROKER_ID=0 -e KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181 -e KAFKA_LISTENERS=PLAINTEXT://127.0.0.1:9092 --link zookeeper:zookeeper -t wurstmeister/kafka:2.12-2.1.1
   2. 命令解析一下，
      1. --link：链接另一个容器，参数格式为：目标容器名称：在本容器内的名称
      2. -e：配置容器内环境变量
   3. 很鸡肋的一个问题，kafka的命令行创建topic的脚本在/opt/kafka_20.1..../目录下的bin/，劳资找了半天
   4. 创建topic：
      1. 首选开启自动创建：auto.create.topics.enable=true
      2. 手动创建：kafka-topic.sh --zookeeper ip：prot --create --topic topicname --partitions 1 --replication-factor 1 --config max.message.bytes=64000 --config flush.messages=1
      3. 命令解析
         1. --zookeeper： 关联zookeeper，可选zookeeper地址： 可选zk集群，ip1：9092/mnt/../...，ip2：9092 逗号分隔
         2. --partition ：指定分区数，这个参数需要根据broker数量决定
         3. --replication-factor：指定partition的replicas数，官网建议2
3. docker ps 观察一下就好，至于clickhouse的启动命令，之前的博客创建过，这里不多做描述