【clickhouse踩坑记录】kafka engine构造实时数据流

前段时间看了字节跳动内部技术沙龙分享，利用kafka engine构造的实时数据架构。故利用现有的资源，整起来。

1. kafka engine的使用，常用架构如下：kafka engine表+materialized view+ ReplicatedReplacingMergeTree的形式。

kafka engine表:消费kafka数据，保存着最原始的数据格式。
ReplicatedReplacingMergeTree表：合并树表，用来存储ods层数据。
materialized view（物化视图）：连接kafka engine表跟ods层的桥梁。

2. kafka消息体如下：

{
   "data":{
   		"order_id":"0001",
   		"update_time":"2021-01-01 00:00:00"
   	},
   "modify_time":"2021-01-01 00:00:00"
}

3. 建表
   因为kafka消息体中，含有嵌套的json，所以kafka Engine表并没有以JSONEachRow进行分割，而是采用了TabSeparated。如果用JSONEachRow，内部的json内容存不了。

use tmp_db;
-- kafka引擎表
CREATE TABLE order_info_kafka (
`message` String
) ENGINE = Kafka('${ip}:${host},${ip}:${host}', '${kafa_topic}', '${groupId}') 
SETTINGS kafka_format = 'TabSeparated'
,kafka_num_consumers = 4

-- mergetree引擎表
CREATE TABLE order_info_d_mt (
, data String
, `event_date` String
, `order_id` String
,  update_time String
,  modify_time String
, `version_id` String
) ENGINE = ReplicatedReplacingMergeTree('/clickhouse/{cluster}/tmp_db/order_info_d_mt/{shard}', '{replica}', version_id) 
PARTITION BY event_date 
PRIMARY KEY (event_date, order_id) 
ORDER BY (event_date, order_id) 
SETTINGS index_granularity = 8192

-- 物化视图
CREATE materialized view if not exists order_info_d_view to order_info_d_mt
as select JSONExtractRaw(message,'data') as data
, substring(JSONExtractString(message,'modify_time'),1,10) as event_date
, JSONExtractString(data,'order_id') as order_id
, JSONExtractString(data,'update_time') as update_time
, JSONExtractString(message,'modify_time') as modify_time
, case when modify_time = '' then '0' else replaceRegexpOne(modify_time,'(\d{4})-(\d{2})-(\d{2}) (\d{2}):(\d{2}):(\d{2})','\1\2\3\4\5\6') end as version_id
from order_info_d_kafka 

4. 至此，实时数据从kafka->clickhouse的ods，就跑通了。