如何在KSQL中基于含JSON数组的Kafka Topic创建表？

核心配置说明 #

1. value_format参数：因为你的Topic消息是JSON格式，直接设置为'JSON'即可。
2. 表结构定义：消息是包含JSON对象的数组，需要用ARRAY<STRUCT<...>>类型定义列，把数组内每个对象的字段映射为STRUCT的属性。另外KSQL的TABLE必须指定主键，若你的Kafka消息没有天然主键字段，可使用ROWKEY作为虚拟主键（对应Kafka消息的键）。

完整CREATE TABLE语句 #

CREATE TABLE test (
  events ARRAY<STRUCT<
    name STRING,
    type STRING,
    num_rep INTEGER
  >>,
  ROWKEY STRING PRIMARY KEY
) WITH (
  KAFKA_TOPIC='testtopic',
  VALUE_FORMAT='JSON',
  KEY_FORMAT='KAFKA' -- 若Kafka消息的键为其他格式可调整，比如'JSON'
);

额外建议：用STREAM处理更灵活 #

如果你的场景是流式分析（无需主键关联、仅消费处理数据），使用STREAM更合适，不需要强制指定主键：

CREATE STREAM test_stream (
  events ARRAY<STRUCT<
    name STRING,
    type STRING,
    num_rep INTEGER
  >>
) WITH (
  KAFKA_TOPIC='testtopic',
  VALUE_FORMAT='JSON'
);

拆分数组查询数据 #

数组类型的数据直接查询会返回完整数组，若要拆分出单个元素，可使用UNNEST函数：

SELECT 
  event.name,
  event.type,
  event.num_rep
FROM test_stream
CROSS JOIN UNNEST(events) AS t(event);

内容的提问来源于stack exchange，提问作者Dasha