使用 STRUCT 类型的 KSQL 是否支持在 JSON 中定义的嵌套对象？答案

【问题标题】：Are nested objects defined in JSON supported in KSQL using the STRUCT type?使用 STRUCT 类型的 KSQL 是否支持在 JSON 中定义的嵌套对象？
【发布时间】：2020-01-09 20:01:44
【问题描述】：

如何创建一个 KSQL 流监听 TOPIC T，其中消息的 JSON 结构是：

{"k":"1","a":{"b":1,"c":{"d":10}}}

我尝试了以下方法，但它不起作用。出现语法错误。

create stream s (k VARCHAR,a STRUCT <b INT ,c <STRUCT d INT >> ) 
                 with (KAFKA_TOPIC='T',VALUE_FORMAT='JSON',KEY='k')

【问题讨论】：

标签： apache-kafka ksqldb confluent-platform

【解决方案1】：

这是一个如何做的例子。我有一个主题的测试数据：

ksql> PRINT test FROM BEGINNING;
Format:JSON
{"ROWTIME":1578571011016,"ROWKEY":"null","k":"1","a":{"b":1,"c":{"d":10}}}
^CTopic printing ceased

声明流：

ksql> CREATE STREAM TEST (K VARCHAR, 
                          A STRUCT<B INT,
                                   C STRUCT<D INT>>
                          ) WITH (KAFKA_TOPIC='test', 
                                  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 * FROM TEST EMIT CHANGES LIMIT 1;
+---------------------+----------+-----+---------------------+
|ROWTIME              |ROWKEY    |K    |A                    |
+---------------------+----------+-----+---------------------+
|1578571011016        |null      |1    |{B=1, C={D=10}}      |
Limit Reached
Query terminated

ksql> SELECT K, A, A->B, A->C, A->C->D FROM TEST EMIT CHANGES LIMIT 1;
+----+-----------------+-------+---------+----------+
|K   |A                |A__B   |A__C     |A__C__D   |
+----+-----------------+-------+---------+----------+
|1   |{B=1, C={D=10}}  |1      |{D=10}   |10        |
Limit Reached
Query terminated

【讨论】：