部署

1. zookeeper的部署
2. kafka部署
3. clickhouse部署

建立联系

联系是在clickhouse进行建表操作实现。

创建消费kafka queue表

	CREATE TABLE xxxtablexxx_queue 
	(
		q_date Date,
		q_msg String
	) ENGINE = Kafka SETTINGS
	kafka_broker_list = 'localhost:9092',
	kafka_topic_list = 'topicname',
	kafka_group_name = 'groupname',
	kafka_format = 'JSONEachRow',
	kafka_row_delimiter = '\n',
	kafka_skip_broken_messages = 1;

创建Clickhouse实际存储数据表

	CREATE TABLE xxxtablexxx_table
	(
		date Date,
		msg String
	) ENGINE = MergeTree()
	ORDER BY date;

创建kafka消费队列和数据表映射表

CREATE MATERIALIZED VIEW xxxtablexxx_consumer to xxxtablexxx_table as select q_date as date, q_msg as msg from dbname.xxxtablexxx_queue;

注意点

场景1:

现象：
当kafka生产者发布一条不符合clickhouse物化表的格式数据时，后续正确发布的消息也无法入库。
解决方案：
在创建队列表中加入kafka_skip_broken_messages = 1字段；