ClickHouse 实战笔记 第01期：Kafka 数据同步到 ClickHouse

ClickHouse 实战笔记 第01期：Kafka 数据同步到 ClickHouse

马听 悦专栏

做者简介

马听，多年 DBA 实战经验，对 MySQL、 Redis、ClickHouse 等数据库有必定了解，专栏《一线数据库工程师带你深刻理解 MySQL》、《Redis 运维实战》做者。

从这一节开始，将经过几期跟各位分享个人一些 ClickHouse 实战笔记。
这一期首先聊聊 Kafka 数据同步到 ClickHouse 的其中一个方案：经过 Kafka 引擎方式同步，下面进入实际操做过程（环境：CentOS7.4）：

1 Kafka 基础环境搭建

由于主要是为了测试数据同步，所以 Kafka 只简单安装了单机版本。

1.1 安装 JDK

cd /usr/src

在这里[https://www.oracle.com/java/technologies/javase/javase-jdk8-downloads.html]选择合适的 JDK 版本，并下载。

tar zxvf jdk-8u261-linux-x64.tar.gz
mv jdk1.8.0_261/ java

编辑 /etc/profile

vim /etc/profile

加入如下内容：

JAVA_HOME=/usr/src/java
PATH=$PATH:$JAVA_HOME/bin
export JAVA_HOME PATH

执行

source /etc/profile

1.2 安装 kafka

cd /usr/src

在这里[http://archive.apache.org/dist/kafka/2.0.0/]选择合适的 kafka 版本，并下载。

tar zxvf kafka_2.11-2.0.0.tgz
mv kafka_2.11-2.0.0 kafka

1.3 启动 zk

cd /usr/src/kafka

./bin/zookeeper-server-start.sh -daemon config/zookeeper.properties

1.4 启动 kafka

nohup ./bin/kafka-server-start.sh config/server.properties &

1.5 建立 topics

./bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

1.6 查看 topics

./bin/kafka-topics.sh --list --zookeeper localhost:2181

1.7 产生消息

./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

执行完上面命令后，会出现下面的窗口：

而后在 > 后面输入须要产生的消息，以下：

1.8 消费消息

另外开一个链接窗口，执行：

cd /usr/src/kafka/
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning

能够看到在 1.7 步骤生成的消息。

2 安装 ClickHouse

ClickHouse 单机版安装参考：https://clickhouse.tech/docs/zh/getting-started/install/

3 建立消费表

在 ClickHouse 上建立 kafka 消费表
登陆 ClickHouse

clickhouse-client

进行建库建表操做：

create database kafka_data;

use kafka_data;

create table kafka_queue(id UInt32,code String,name String)engine =Kafka() settings kafka_broker_list = 'localhost:9092',kafka_topic_list='test',kafka_group_name='group1',kafka_format='JSONEachRow',kafka_skip_broken_messages=100;

注：

• kafka_broker_list：kafka 的链接地址和端口。
• kafka_topic_list：kafka 的 topic 名。
• kafka_group_name：kafka 的组名。
• kafka_format：表示用于解析消息的数据格式，消息发送端必须按此格式发送消息。
• kafka_skip_broken_messages：当解析数据出现错误时，运行跳过失败的数据行数。

4 建立存储表

由于 Kafka 消费表不能直接做为结果表使用。Kafka 消费表只是用来消费Kafka数据，没有真正的存储全部数据，只要查询一次，数据就会清空。所以须要在 ClickHouse 中建立存储表保存数据。
在 ClickHouse 上建立存储表：

create table kafka_table(id UInt32,code String,name String) engine=MergeTree() order by id

5 建立数据同步视图

建立 view 把 kafka 消费表消费到的数据导入 ClickHouse 存储表：

create materializedview consumer to kafka_table as select id,code,name from kafka_queue

6 测试数据同步

/usr/src/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

输入：

{"id":2,"code":"two","name":"aa"}

确认 ClickHouse 存储表是否能正常获取到数据

select * from kafka_table

7 其余维护操做

中止数据同步，能够删除视图

drop table consumer

或者卸载

detach table consumer

卸载以后，若是想再次恢复，可使用：

attach materialized view consumer to kafka_table(id UInt32,code String,name String)as select id,code,name from kafka_queue

8 存在的问题

经过 Kafka 引擎进行数据同步的方式尽管很方便，可是在实战过程当中发现，Kafka 吐出来的数据不必定会是 {"id":2,"code":"two","name":"aa"} 这类格式，这种状况能够考虑使用另一种方案：借助 Flume 实现 Kafka 到 CH 的同步，这个方案将在后续的文章中进行介绍。