场景:
从 kafka 消费原始报文,做简单的数据抽取,并把原文存储到 doris 表中
表结构:
[ol]
[/ol]
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 "报文导入";
ROUTINE LOAD 可以正常创建,但是运行起来后就会报错:
Reason: [DATA_QUALITY_ERROR]empty json path. src line [];
如果把 jsonpaths 改为"["$.k1","$.k2"]"
就能正常写入数据
但是业务场景需要将 kafka 消息原文进行存储,请问大佬应该如何处理