时间戳

首页 > 解决方案 > 卡住运动交错窗口

amazon-web-services - 卡住运动交错窗口

问题描述

我设置了一个运动分析应用程序，它从具有以下架构的运动流中获取数据。

--------------------------
Column          ColumnType
--------------------------
Level           varchar(10)
RootID          varchar(32)
ProcessID       varchar(16)
EntityName      varchar(64)
Message         varchar(512)
Threshold       varchar(32)
TriggerTime     timestamp

我的目标是创建一个实时运动分析解决方案，将记录与“溢出”级别隔离，并根据 RootID 对它们进行分组。理想情况下，属于 RootID 的所有记录都应在 5 分钟内到达 kinesis。所以我正在考虑为此设置一个交错窗口，到目前为止我已经想出了这个 SQL。

CREATE OR REPLACE STREAM "OVERFLOW_SQL_STREAM" (
    "Level" varchar (10),
    "RootID" varchar (32),
    "ProcessID" varchar(16),
    "EntityName" varchar(64),
    "Message" varchar(512),
    "Threshold" varchar(32),
    "TriggerTime" timestamp
);
CREATE OR REPLACE PUMP "STREAM_PUMP" AS
    INSERT INTO "OVERFLOW_SQL_STREAM"
        SELECT STREAM 
            "Level" varchar (10),
            "RootID" varchar (32),
            "ProcessID" varchar(16),
            "EntityName" varchar(64),
            "Message" varchar(512),
            "Threshold" varchar(32),
            "TriggerTime" timestamp
        FROM "SOURCE_SQL_STREAM_001"
        WHERE "Level" like "OVERFLOW"
        WINDOWED BY STAGGER (
            PARTITION BY "RootID",FLOOR("TriggerTime" TO MINUTE) RANGE INTERVAL '5' MINUTE);

我在 SQL 中收到一条错误消息，指出“PARTITION BY 子句没有列 'Level'”。我不明白为什么要将该列添加到分区，因为我希望我的记录仅由 RootID 列而不是任何其他列进行分区。添加该列会引发错误，提示我应该添加下一列，依此类推。我无法理解错误。请帮助我！谢谢！

标签： amazon-web-servicesamazon-kinesis

解决方案

此类问题有一种解决方法。您可以使用 FIRST_VALUE() 或 LAST_VALUE() 来转换结果，而不是直接传递它们。

CREATE OR REPLACE PUMP "STREAM_PUMP" AS
    INSERT INTO "OVERFLOW_SQL_STREAM"
    SELECT STREAM 
        LAST_VALUE("Level") AS Level,
       "RootID" varchar (32),
       ....
       ....
       ....
       "TriggerTime" timestamp
    FROM "SOURCE_SQL_STREAM_001"
    WHERE "Level" like "OVERFLOW"
    WINDOWED BY STAGGER (
        PARTITION BY "RootID",FLOOR("TriggerTime" TO MINUTE) RANGE INTERVAL '5' MINUTE);

这是您无需添加到 PARTITION BY 子句即可创建流泵的方式。

FIRST_VALUE() -- 获取与流分区匹配的级别的第一个值（此处为 RootID）
LAST_VALUE() --反之亦然

推荐阅读