代码之家 › 专栏 › 技术社区 › Kirzilla

包含原始JSON消息部分的ksql表或流中的字符串字段

ksql apache-kafka

0

Kirzilla · 技术社区 · 5 年前

是否可以将字符串字段添加到包含原始消息JSON一部分的ksql表/流中?

例如,

原始邮件:

{userId:12345, 
 service:"service-1", 
 "debug":{
          "msg":"Debug message", 
          "timer": 11.12}
}

所以,我们需要绘制地图 userId 到 userId BIGINT , service 到 service STRING 和 debug 到 debug STRING 其中将包含 {"msg":"Debug message", "timer": 11.12} 作为字符串。

1 回复 | 直到 5 年前

1

Robin Moffatt 5 年前

是的,您可以简单地将其声明为 VARCHAR . 从这里,您可以将其视为恰好是JSON的字符串,或者您可以使用 EXTRACTJSONFIELD 功能。

将示例消息发送到主题:

echo '{"userId":12345, "service":"service-1", "debug":{ "msg":"Debug message", "timer": 11.12} }' | kafkacat -b localhost:9092 -t test_topic -P

声明流:

ksql> CREATE STREAM demo (userid BIGINT, service VARCHAR, debug VARCHAR) WITH (KAFKA_TOPIC='test_topic', VALUE_FORMAT='JSON');

 Message
----------------
 Stream created
----------------

查询列:

ksql> SET 'auto.offset.reset' = 'earliest';
Successfully changed local property 'auto.offset.reset' to 'earliest'. Use the UNSET command to revert your change.
ksql> SELECT USERID, SERVICE, DEBUG FROM demo;
12345 | service-1 | {"msg":"Debug message","timer":11.12}

访问嵌套的JSON字段:

ksql> SELECT USERID, SERVICE, EXTRACTJSONFIELD(DEBUG,'$.msg') FROM demo;
12345 | service-1 | Debug message

ksql> SELECT USERID, SERVICE, EXTRACTJSONFIELD(DEBUG,'$.timer') FROM demo;
12345 | service-1 | 11.12