Stagger Windows

ずらしウィンドウは、一貫性のない時間に届くデータのグループを分析するのに適したウィンドウ処理メソッドです。これは、関連する一連のセールスやログレコードなど、時系列分析のユースケースに適しています。

たとえば、VPC フローログには約 10 分のキャプチャウィンドウがあります。しかし、クライアントにデータを集約する場合は最大 15 分のキャプチャウィンドウを持つことができます。ずらしウィンドウは、これらのログを分析のために集計するのに理想的です。

ずらしウィンドウでは、タンブリングウィンドウが使用されたときなど、同じ時間制限付きウィンドウに収まらない関連レコードの問題を解決します。

Tumbling Windows の部分的な結果

遅延データまたは順序通りでないデータの集約に タンブリングウィンドウ を使用する場合、一定の制限があります。

時間関連のデータのグループを分析するためにタンブリングウィンドウを使用する場合、個々のレコードは別々のウィンドウに分類される可能性があります。したがって、各ウィンドウの部分的な結果を後で組み合わせて、各レコードグループの完全な結果を得る必要があります。

次のタンブリングウィンドウクエリでは、レコードは行時間、イベント時間、およびティッカーシンボルによってウィンドウにグループ化されます。

CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" (
    TICKER_SYMBOL VARCHAR(4),
    EVENT_TIME timestamp,
    TICKER_COUNT     DOUBLE);

CREATE OR REPLACE PUMP "STREAM_PUMP" AS 
  INSERT INTO "DESTINATION_SQL_STREAM" 
    SELECT STREAM 
        TICKER_SYMBOL,
        FLOOR(EVENT_TIME TO MINUTE),
        COUNT(TICKER_SYMBOL) AS TICKER_COUNT
    FROM "SOURCE_SQL_STREAM_001"
    GROUP BY ticker_symbol, FLOOR(EVENT_TIME TO MINUTE), STEP("SOURCE_SQL_STREAM_001".ROWTIME BY INTERVAL '1' MINUTE);

次の図でアプリケーションは、1 分の粒度で取引が発生したとき (イベント時間) に基づき、受信した取引の数をカウントしています。アプリケーションは行時間とイベント時間に基づき、タンブリングウィンドウを使用してデータをグループ化できます。アプリケーションは、すべてが 1 分間隔で届く 4 つのレコードを受け取ります。次に、行時間、イベント時間、およびティッカーシンボルでレコードをグループ化します。レコードの一部は最初のタンブリングウィンドウが終了してから届くため、すべてのレコードが同じ 1 分のタンブリングウィンドウに収まるわけではありません。

前述の図には、以下のイベントが含まれます。

ROWTIME	EVENT_TIME	TICKER_SYMBOL
11:00:20	11:00:10	AMZN
11:00:30	11:00:20	AMZN
11:01:05	11:00:55	AMZN
11:01:15	11:01:05	AMZN

タンブリングウィンドウアプリケーションからの結果セットは、以下のようになります。

ROWTIME	EVENT_TIME	TICKER_SYMBOL	COUNT
11:01:00	11:00:00	AMZN	2
11:02:00	11:00:00	AMZN	1
11:02:00	11:01:00	AMZN	1

前述の結果では、3 つの結果が返されます。

• 最初の 2 つのレコードを集計する、ROWTIME が 11:01:00 のレコード。

• 3 つ目のレコードのみを集計する、11:02:00 のレコード。このレコードは、2 番目のウィンドウ内に ROWTIME がありますが、EVENT_TIME は 1 番目のウィンドウ内にあります。

• 4 つ目のレコードのみを集計する、11:02:00 のレコード。

完全な結果セットを分析するには、レコードが永続的なストアに集約されている必要があります。これにより、アプリケーションに複雑性と処理要件が加わります。

Stagger Windows での完全な結果

時間関連のデータレコードの分析の精度を向上させるため、Kinesis Data Analytics ではずらしウィンドウという新しいウィンドウのタイプを提供しています。このウィンドウタイプでは、パーティションキーに一致する最初のイベントが届いたときにウィンドウが開きます。固定の時間間隔でウィンドウが開くことはありません。ウィンドウは、ウィンドウを開いたときから測定される、指定された経過時間に基づいて閉じます。

ずらしウィンドウは、ウィンドウ句の各キーグループのための、別個の時間制限付きウィンドウです。アプリケーションは、すべての結果に対して単一のウィンドウを使用するのではなく、独自の時間ウィンドウ内にウィンドウ句のそれぞれの結果を集計します。

次のずらしウィンドウクエリでは、レコードはイベント時間、およびティッカーシンボルによってウィンドウにグループ化されます。

CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" (
    ticker_symbol    VARCHAR(4), 
    event_time       TIMESTAMP,
    ticker_count     DOUBLE);

CREATE OR REPLACE PUMP "STREAM_PUMP" AS 
  INSERT INTO "DESTINATION_SQL_STREAM" 
    SELECT STREAM 
        TICKER_SYMBOL,
        FLOOR(EVENT_TIME TO MINUTE),
        COUNT(TICKER_SYMBOL) AS ticker_count
    FROM "SOURCE_SQL_STREAM_001"
    WINDOWED BY STAGGER (
            PARTITION BY FLOOR(EVENT_TIME TO MINUTE), TICKER_SYMBOL RANGE INTERVAL '1' MINUTE);
        
        

次の図では、イベントはイベント時間、およびティッカーシンボルによってずらしウィンドウに集計されます。

前述の図には、タンブリングウィンドウアプリケーションが分析したものと同じイベントである、以下のイベントが含まれています。

ROWTIME	EVENT_TIME	TICKER_SYMBOL
11:00:20	11:00:10	AMZN
11:00:30	11:00:20	AMZN
11:01:05	11:00:55	AMZN
11:01:15	11:01:05	AMZN

ずらしウィンドウアプリケーションからの結果セットは、以下のようになります。

ROWTIME	EVENT_TIME	TICKER_SYMBOL	Count
11:01:20	11:00:00	AMZN	3
11:02:15	11:01:00	AMZN	1

返されたレコードは、最初の 3 つの入力レコードを集計します。レコードは、1 分間のずらしウィンドウでグループ化されます。ずらしウィンドウは、アプリケーションが最初の AMZN レコード (ROWTIME が 11:00:20 のもの) を受信したときに開始されます。1 分間のずらしウィンドウが終了すると (11:01:20)、ずらしウィンドウ内に収められる結果 (ROWTIME および EVENT_TIME に基づく) が、出力ストリームに書き込まれます。ずらしウィンドウを使用すると、1 分間ウィンドウ内にある ROWTIME および EVENT_TIME を持つすべてのレコードが 1 つの結果として出力されます。

最後のレコード (1 分間集計から外れた EVENT_TIME がある) は別々に集計されます。これは、レコードを結果セットに分割するために使用されるパーティションキーの 1 つが EVENT_TIME であり、最初のウィンドウの EVENT_TIME のパーティションキーが 11:00 であるためです。

ずらしウィンドウの構文は、WINDOWED BY という特別な句で定義されています。この句は、ストリーミング集計の GROUP BY 句の代わりに使用されます。この句は、オプションの WHERE 句の直後、および HAVING 句の前に表示されます。

ずらしウィンドウは、WINDOWED BY 句で定義され、パーティションキーとウィンドウ長の 2 つのパラメータを取ります。パーティションキーは、受信データストリームを分割し、ウィンドウが開いたときに定義します。ずらしウィンドウは、固有のパーティションキーを持つ最初のイベントがストリームに表示されたとき開きます。ずらしウィンドウは、ウィンドウ長により定義された一定期間の後で閉じます。次のコード例にその構文を示します。

...
FROM <stream-name>
WHERE <... optional statements...>
WINDOWED BY STAGGER(
	PARTITION BY <partition key(s)>
	RANGE INTERVAL <window length, interval>
);