如何在Amazon Kinesis Analytics SQL中将窗口列值聚合为数组输出

你使用的AWS Kinesis Data Analytics SQL runtime内置支持LISTAGG聚合函数，可以直接实现需求，无需自定义扩展逻辑，修改后的完整SQL如下：

CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" (
    "event_type" varchar(32),
    "count_products_viewed" integer,
    "max_price_viewed" decimal(7,2),
    "list_products_viewed" varchar(2048) -- 长度可根据单窗口最大商品数量自行调整
);

CREATE OR REPLACE PUMP "STREAM_PUMP" AS 
    INSERT INTO "DESTINATION_SQL_STREAM"
        SELECT STREAM 
            "stream_1"."event",
            COUNT("stream_1"."product_id") as "count_products_viewed",
            MAX("stream_1"."product_price") as "max_price_viewed",
            -- 直接拼接得到你需要的[1111, 2222]格式输出
            '[' || LISTAGG("stream_1"."product_id", ',') || ']' as "list_products_viewed"
        FROM "SOURCE_SQL_STREAM_001" as "stream_1"
        WINDOWED BY STAGGER (
            PARTITION BY
            "stream_1"."event"
            RANGE INTERVAL '30' MINUTE
        );

补充说明 #

• 若需要排除重复的商品ID，可将LISTAGG("stream_1"."product_id", ',')替换为LISTAGG(DISTINCT "stream_1"."product_id", ',')
• 若需要标准可解析的JSON格式（ID带双引号），可将该字段逻辑替换为'[' || LISTAGG('"' || "stream_1"."product_id" || '"', ',') || ']'，输出结果为["1111","2222"]，可直接被JSON解析器识别

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