list - ksql中的可变长度列表
问题描述
在 KSQL 中,可以EXTRACTJSONFIELD
用于嵌套结构,但我不知道如何处理可变长度列表。例如:
{"id":1,"quux":[{"x":1,"y":2},{"x":3,"y":4},{"x":5,"y":6}]}
我可以quux
作为基本流的 varchar 处理,
create stream mystream (id bigint, quux varchar)
with (kafka_topic='mytopic', value_format='json')
但我希望能够把它变成一张桌子:
quuxid x y
1 1 2
1 3 4
1 5 6
如何处理 KSQL 中的可变长度列表?
解决方案
这在 KSQL 中目前是不可能的。
如您所见,您可以通过索引专门访问数组数据:填充测试数据:
echo '{"id":1,"quux":[{"x":1,"y":2},{"x":3,"y":4},{"x":5,"y":6}]}' | \
kafkacat -b localhost:9092 -t quux
检查 KSQL 中的消息:
ksql> print 'quux' from beginning;
Format:JSON
{"ROWTIME":1528791985250,"ROWKEY":"null","id":1,"quux":[{"x":1,"y":2},{"x":3,"y":4},{"x":5,"y":6}]}
创建流:
create stream mystream (id bigint, quux varchar) \
with (kafka_topic='quux', value_format='json');
查询流:
ksql> SET 'auto.offset.reset' = 'earliest';
Successfully changed local property 'auto.offset.reset' from 'earliest' to 'earliest'
ksql>
ksql> select * from mystream;
1528791985250 | null | 1 | [{"x":1,"y":2},{"x":3,"y":4},{"x":5,"y":6}]
创建流array
:
ksql> CREATE STREAM mystream2 (id bigint, quux array<varchar>) with (kafka_topic='quux', value_format='json');
Message
----------------
Stream created
----------------
按索引访问单个项目:
ksql> SELECT quux[0], EXTRACTJSONFIELD(quux[0],'$.x') AS X, EXTRACTJSONFIELD(quux[0],'$.y') AS Y from mystream2;
{"x":1,"y":2} | 1 | 2
但是您正在寻找一个EXPLODE
函数的等价物,它在 KSQL 中尚不存在。
相关的github问题:
推荐阅读
- sql-server - 试图在未按顺序包含的列上获得平局结果
- r - 从 R 中的日期集中获取工作日
- java - java - 如何计算我的程序在java中每秒执行的次数或读取操作?
- nestjs - “用户”类型上不存在属性“checkPassword”
- java - 仅当满足特定条件时,如何激活 OSGI 包?
- c# - OnApplicationPause 延迟工作
- c++ - 抛出未处理的异常:读取访问冲突。_Pnext 为 0xC
- reactjs - 创建用于拆分字符串的自定义 Hook (React JS)
- python - 函数的monkeypatching中的setattr抛出属性错误
- angular - 递归调用角组件时,mat-accordion 滞后