创建KSQL流：如何从复杂JSON中提取字段值

我来帮你搞定这个KSQL提取数组字段的问题——你之前踩的坑主要是没充分利用KSQL的复杂类型支持，还搞错了数组索引的规则，下面一步步给你讲正确的做法：

1. 先正确定义包含复杂类型的流 #

你的JSON里used_service_units是一个嵌套结构体的数组，KSQL原生支持ARRAY<STRUCT<>>这种复杂类型，不需要把数组转成字符串处理。我们先把流的结构定义准确：

CREATE STREAM usage (
    event_type VARCHAR,
    created_at VARCHAR,
    used_service_units ARRAY<STRUCT<
        amount BIGINT,
        currency VARCHAR,
        unit_of_measure VARCHAR
    >>
) WITH (
    kafka_topic='usage_events',
    value_format='json'
);

注意：示例里的amount是整数2412739，用BIGINT更匹配；如果是小数可以换成DOUBLE。

2. 提取数组单个元素的字段 #

KSQL的数组是1-based索引（不是你习惯的0-based），所以要取第一个元素的amount，直接用索引访问即可：

SELECT 
    event_type,
    created_at,
    used_service_units[1].amount AS service_amount
FROM usage;

这条语句就能直接返回你想要的第一个数组元素的amount值。

3. 处理数组多个元素（展开成多行） #

如果used_service_units可能包含多个元素，想要逐个处理每个元素的amount，可以用EXPLODE函数把数组展开成独立的行：

第一步：创建展开后的流 #

CREATE STREAM usage_units_expanded AS
SELECT 
    event_type,
    created_at,
    EXPLODE(used_service_units) AS service_unit
FROM usage;

第二步：查询展开后的字段 #

SELECT 
    event_type,
    created_at,
    service_unit.amount AS service_amount,
    service_unit.unit_of_measure AS service_unit_type
FROM usage_units_expanded;

这样数组里的每个元素都会变成单独的一行记录，方便后续做聚合或其他处理。

为什么你之前的方法失败了？ #

• 方法1错误原因：流定义只能声明字段的类型，不能直接在字段名里写数组索引（比如used_service_units[0].amount），KSQL会把[识别成语法错误。
• 方法2错误原因：你把used_service_units定义成了VARCHAR，但KSQL已经能自动解析JSON数组为原生数组类型，转成字符串反而破坏了结构；再加上KSQL数组是1-based索引，用[0]会触发索引越界，导致代码生成失败。

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