ksql 中的可变长度列表

Question

在 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')

但我希望能够通过以下方式将其变成 table：

如何处理 KSQL 中的可变长度列表？

Answer 1

这在 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 中尚不存在。

ksql 中的可变长度列表

variable-length lists in ksql

list

apache-kafka

ksqldb