【clickhouse踩坑记录】kafka engine构造实时数据流
背景前段时间看了字节跳动内部技术沙龙分享,利用kafka engine构造的实时数据架构。故利用现有的资源,整起来。实践过程kafka engine的使用,常用架构如下:kafka engine表+materialized view+ReplicatedReplacingMergeTree的形式。kafka engine表:消费kafka数据,保存着最原始的数据格式。ReplicatedRepla
·
背景
前段时间看了字节跳动内部技术沙龙分享,利用kafka engine构造的实时数据架构。故利用现有的资源,整起来。
实践过程
- kafka engine的使用,常用架构如下:kafka engine表+materialized view+ ReplicatedReplacingMergeTree的形式。
kafka engine表:消费kafka数据,保存着最原始的数据格式。
ReplicatedReplacingMergeTree表:合并树表,用来存储ods层数据。
materialized view(物化视图):连接kafka engine表跟ods层的桥梁。
- kafka消息体如下:
{
"data":{
"order_id":"0001",
"update_time":"2021-01-01 00:00:00"
},
"modify_time":"2021-01-01 00:00:00"
}
- 建表
因为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
- 至此,实时数据从kafka->clickhouse的ods,就跑通了。
更多推荐
已为社区贡献2条内容
所有评论(0)