场景: 从 kafka 消费原始报文,做简单的数据抽取,并把原文存储到 doris 表中
表结构:
routine load 配置
CREATE ROUTINE LOAD xxx.yyy
ON zzz
COLUMNS (
k1 ,
k2 ,
packet
)
PROPERTIES
(
"format" = "json",
"jsonpaths"="[\"$.k1\",\"$.k2\",\"$\"]",
"max_batch_rows"="1000000",
"max_batch_size"="104857600",
"max_filter_ratio"="0.5",
"max_error_number" = "1000",
"strict_mode" = "true"
)
FROM KAFKA
(
"kafka_broker_list" = "xxx",
"kafka_topic" = "xxx",
"property.kafka_default_offsets" = "OFFSET_BEGINNING",
"property.group.id" = "doris-routine"
)
COMMENT "报文导入";
如果把 jsonpaths 改为"["$.k1","$.k2"]" 就能正常写入数据
但是业务场景需要将 kafka 消息原文进行存储,请问大佬应该如何处理
1
dreamxj001 OP 消息报文结构是:
|
2
dreamxj001 OP @dreamxj001 {"k1":"111", "k2":"222", "k3":"333",.....}
|