背景

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

实践过程

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

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

  1. kafka消息体如下:
{
   "data":{
   		"order_id":"0001",
   		"update_time":"2021-01-01 00:00:00"
   	},
   "modify_time":"2021-01-01 00:00:00"
}
  1. 建表
    因为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 

  1. 至此,实时数据从kafka->clickhouse的ods,就跑通了。
Logo

Kafka开源项目指南提供详尽教程,助开发者掌握其架构、配置和使用,实现高效数据流管理和实时处理。它高性能、可扩展,适合日志收集和实时数据处理,通过持久化保障数据安全,是企业大数据生态系统的核心。

更多推荐