慎重に検討した結果、Amazon Kinesis Data Analytics for SQL アプリケーションを中止することにしました。 1. **2025 年 9 月 1** 日以降、Amazon Kinesis Data Analytics for SQL アプリケーションのバグ修正は提供されません。これは、今後の廃止によりサポートが制限されるためです。 2. **2025 年 10 月 15** 日以降、新しい Kinesis Data Analytics for SQL アプリケーションを作成することはできません。 3. **2026 年 1 月 27 日**以降、アプリケーションは削除されます。Amazon Kinesis Data Analytics for SQL アプリケーションを起動することも操作することもできなくなります。これ以降、Amazon Kinesis Data Analytics for SQL のサポートは終了します。詳細については、「[Amazon Kinesis Data Analytics for SQL アプリケーションのサポート終了](discontinuation.md)」を参照してください。 翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。 # ストリーミング SQL の概念 Amazon Kinesis Data Analytics は、ANSI 2008 SQL 標準を拡張機能とともに実装しています。これらの拡張機能により、ストリーミングデータを処理できます。以下のトピックで、キーとなるストリーミング SQL の概念を取り上げます。 **Topics** + [アプリケーション内ストリームとポンプ](streams-pumps.md) + [タイムスタンプと ROWTIME 列](timestamps-rowtime-concepts.md) + [連続クエリ](continuous-queries-concepts.md) + [ウィンドウクエリ](windowed-sql.md) + [ストリーミングデータオペレーション: ストリーム結合](stream-joins-concepts.md) # アプリケーション内ストリームとポンプ [アプリケーション入力](https://docs.aws.amazon.com/kinesisanalytics/latest/dev/how-it-works-input.html)を設定する際に、ストリーミングソースを作成済みのアプリケーション内ストリームにマッピングします。データは絶えずストリーミングソースからアプリケーション内ストリームに流れます。アプリケーション内ストリームは、SQL ステートメントを使用してクエリできるテーブルのように機能しますが、データが絶えず流れているためにストリームと呼ばれます。 **注記** アプリケーション内ストリームを Amazon Kinesis データストリームや Firehose 配信ストリームと混同しないでください。アプリケーション内ストリームは、Amazon Kinesis Data Analytics アプリケーションのコンテキスト内のみに存在します。Kinesis のデータストリームと Firehose の配信ストリームは、アプリケーションからは独立した存在です。アプリケーションの入力設定におけるストリーミング送信元として、または出力設定における送信先として、これらを設定できます。 また、必要に応じて、中間クエリ結果を保存するためにアプリケーション内ストリームをさらに作成することもできます。アプリケーション内ストリームを作成する手順は、2 ステップです。まず、アプリケーション内ストリームを作成し、そこにデータをポンプします。たとえば、アプリケーションの入力設定で `INPUTSTREAM` という名前のアプリケーション内ストリームを作成するとします。次の例では、別のストリーム (`TEMPSTREAM`) を作成して、`INPUTSTREAM` からそこにデータをポンプしています。 1. 次の図のように、3 つの列を持つアプリケーション内ストリーム (`TEMPSTREAM`) を作成します。 ``` CREATE OR REPLACE STREAM "TEMPSTREAM" ( "column1" BIGINT NOT NULL, "column2" INTEGER, "column3" VARCHAR(64)); ``` 列名は引用符で指定され、大文字小文字を区別します。詳細については、Amazon Kinesis Data Analytics SQL Reference の「[Identifiers](https://docs.aws.amazon.com/kinesisanalytics/latest/sqlref/sql-reference-identifiers.html)」を参照してください。 1. ポンプを使用してストリームにデータを挿入します。ポンプとは、1 つのアプリケーション内ストリームから別のアプリケーション内ストリームにデータを挿入する、連続して実行される挿入クエリです。次のステートメントは、ポンプ (`SAMPLEPUMP`) を作成し、別のストリーム (`INPUTSTREAM`) からレコードを選択して、`TEMPSTREAM` にデータを挿入します。 ``` CREATE OR REPLACE PUMP "SAMPLEPUMP" AS INSERT INTO "TEMPSTREAM" ("column1", "column2", "column3") SELECT STREAM inputcolumn1, inputcolumn2, inputcolumn3 FROM "INPUTSTREAM"; ``` 複数のライターから 1 つのアプリケーション内ストリームに挿入できます。また、ストリームから複数のリーダーを選択できます。アプリケーション内ストリームを、配信/購読メッセージングパラダイムの実装と考えることができます。このパラダイムでは、そこに含まれる作成時刻と受信時刻を含むデータ列はストリーミング SQL ステートメントのカスケードで処理、変換、転送でき、従来の RDBMS に保存する必要はありません。 アプリケーション内ストリームが作成された後は、通常の SQL クエリを実行できます。 **注記** ストリームのクエリを実行するとき、ほとんどの SQL ステートメントは、行ベースまたは時間ベースのウィンドウを使用してバインドされます。詳細については、「[ウィンドウクエリ](windowed-sql.md)」を参照してください。 また、ストリームを結合することもできます。ストリーム結合の例については、「[ストリーミングデータオペレーション: ストリーム結合](stream-joins-concepts.md)」を参照してください。 # タイムスタンプと ROWTIME 列 アプリケーション内ストリームには、`ROWTIME` という特別な行が含まれています。Amazon Kinesis Data Analytics によって最初のアプリケーション内ストリームに行が挿入されると、タイムスタンプが保存されます。`ROWTIME` は、Amazon Kinesis Data Analytics がストリーミングソースからレコードを読み取った後、最初のアプリケーション内ストリームにレコードを挿入した時点のタイムスタンプを反映します。この `ROWTIME` 値はその後、アプリケーション全体で維持されます。 **注記** 1 つのアプリケーション内ストリームから別のアプリケーション内ストリームにレコードをポンプする際に、`ROWTIME` 列を明示的にコピーする必要はありません。この列は Amazon Kinesis Data Analytics でコピーされます。 Amazon Kinesis Data Analytics は、`ROWTIME` の値が一定間隔で増加することを保証します。このタイムスタンプは、時間ベースウィンドウのクエリで使用されます。詳細については、「[ウィンドウクエリ](windowed-sql.md)」を参照してください。 ROWTIME 列には、アプリケーション内ストリームの他の列と同様に、`SELECT` ステートメント内でアクセスできます。例えば、次のようになります。 ``` SELECT STREAM ROWTIME, some_col_1, some_col_2 FROM SOURCE_SQL_STREAM_001 ``` ## ストリーミング分析でのさまざまな時間を理解する `ROWTIME` の他に、リアルタイムストリーミングアプリケーションには別のタイプの時間があります。次のようなものがあります。 + **イベント時間** – イベントが発生したときのタイムスタンプ。*クライアント側の時間*と呼ばれることもあります。イベントが発生した時間であるため、分析でこの時間を使用するのが望ましい場合がよくあります。しかし、携帯電話やウェブクライアントなど多くのイベントソースは信頼性の高い時計を持たないため、時間が不正確になる場合があります。さらに、接続性の問題で、レコードがイベントの発生と同じ順序でストリームに現れない場合があります。   + **取り込み時間** — レコードがストリーミングソースに追加されたときのタイムスタンプ。Amazon Kinesis Data Streams は、このタイムスタンプを提供する `APPROXIMATE_ARRIVAL_TIME` というフィールドをすべてのレコードに含んでいます。*サーバー側の時間*と呼ばれることもあります。取り込み時間は、多くの場合、イベント時間にかなり近い近似値です。ストリームへのレコード取り込みに何らかの遅延が発生した場合は不正確になることがありますが、通常は稀なケースです。また、取り込み時間の順序が入れ替わることはめったにありません。ただし、ストリーミングデータの分散特性のために発生する可能性はあります。そのため、取り込み時間はエベント時間をもっとも正確に順序正しく反映しています。   + **処理時間** — Amazon Kinesis Data Analytics が最初のアプリケーション内ストリームに行を挿入したときのタイムスタンプ。Amazon Kinesis Data Analytics は、このタイムスタンプを各アプリケーション内ストリームに存在する `ROWTIME` 列に提供します。処理時間は常に一定間隔で増加しています。ただし、アプリケーションが遅れている場合は正確ではありません。(アプリケーションが遅れた場合、処理時間がイベント時間を正確に反映しなくなります)。この `ROWTIME` は経過時間に関しては正確ですが、実際にイベントが発生した時間ではない場合があります。 時間ベースのウィンドウクエリでこれらの時間を使用するには、それぞれ利点と欠点があります。これらの時間を 1 つ以上選択し、またそれに伴う欠点に対処する戦略をお客様のユースケースシナリオに基づいて選択することをお勧めします。 **注記** 行ベースのウィンドウを使用する場合は、時刻は問題ではないため、このセクションは無視してかまいません。 `ROWTIME` と他の時間 (取り込み時間またはイベント時間) の 2 つの時間ベースを両方使用した 2 ウィンドウ戦略をお勧めします。 + 次の例に示すように、クエリで結果を発行する頻度を制御する `ROWTIME` を最初のウィンドウとして使用します。論理時間としては使用されません。 + 分析に関連付ける論理時間であるその他の時間のうち 1 つを使用します。この時間は、いつイベントが発生したかを示します。次の例では、分析の目的はレコードをグループ化し、ティッカーでカウントを返すことです。 この戦略の利点は、イベントが発生したときを示す時間を使用できることです。アプリケーションが遅れたときやイベントの到達順序が入れ替わったときに適切に処理できます。アプリケーション内ストリームにレコードを持ってくるときにアプリケーションが遅れた場合でも、2 番目のウィンドウの論理時間でグループ化されます。クエリは `ROWTIME` を使用して処理順序を保証します。遅延したレコード (取り込みタイムスタンプの値が `ROWTIME` 値よりも早い) も正常に処理されます。 [「使用開始」実習](https://docs.aws.amazon.com/kinesisanalytics/latest/dev/get-started-exercise.html)で使用されているデモストリームに対して次のクエリを検討します。クエリは `GROUP BY` 句を使用し、1 分ごとのタンブリングウィンドウでティッカーカウントを発行します。 ``` CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" ("ingest_time" timestamp, "APPROXIMATE_ARRIVAL_TIME" timestamp, "ticker_symbol" VARCHAR(12), "symbol_count" integer); CREATE OR REPLACE PUMP "STREAM_PUMP" AS INSERT INTO "DESTINATION_SQL_STREAM" SELECT STREAM STEP("SOURCE_SQL_STREAM_001".ROWTIME BY INTERVAL '60' SECOND) AS "ingest_time", STEP("SOURCE_SQL_STREAM_001".APPROXIMATE_ARRIVAL_TIME BY INTERVAL '60' SECOND) AS "APPROXIMATE_ARRIVAL_TIME", "TICKER_SYMBOL", COUNT(*) AS "symbol_count" FROM "SOURCE_SQL_STREAM_001" GROUP BY "TICKER_SYMBOL", STEP("SOURCE_SQL_STREAM_001".ROWTIME BY INTERVAL '60' SECOND), STEP("SOURCE_SQL_STREAM_001".APPROXIMATE_ARRIVAL_TIME BY INTERVAL '60' SECOND); ``` `GROUP BY` で、まず 1 分ごとのウィンドウの `ROWTIME` に基づいて、次に `APPROXIMATE_ARRIVAL_TIME` に基づいてレコードをグループ化します。 結果のタイムスタンプ値は、最も近い 60 秒間隔で切り捨てられます。クエリによって発行された最初のグループ結果が、最初の 1 分間のレコードを示しています。発行された 2 つめの結果グループは、`ROWTIME` に基づいた次の分単位のレコードを示しています。最後のレコードは、アプリケーションで、アプリケーション内ストリームにレコードを持ってくるのが後れたことを示します (取り込みタイムスタンプに対して、`ROWTIME` 値が遅れていることを示します)。 ``` ROWTIME INGEST_TIME TICKER_SYMBOL SYMBOL_COUNT --First one minute window. 2016-07-19 17:05:00.0 2016-07-19 17:05:00.0 ABC 10 2016-07-19 17:05:00.0 2016-07-19 17:05:00.0 DEF 15 2016-07-19 17:05:00.0 2016-07-19 17:05:00.0 XYZ 6 –-Second one minute window. 2016-07-19 17:06:00.0 2016-07-19 17:06:00.0 ABC 11 2016-07-19 17:06:00.0 2016-07-19 17:06:00.0 DEF 11 2016-07-19 17:06:00.0 2016-07-19 17:05:00.0 XYZ 1 *** ***late-arriving record, instead of appearing in the result of the first 1-minute windows (based on ingest_time, it is in the result of the second 1-minute window. ``` ダウンストリームデータベースに結果をプッシュすることで、最終的な 1 分あたりの正確なカウントを得るために結果を 1 つにできます。例えば、Amazon Redshift テーブルに書き込む Firehose 配信ストリームに結果を永続化するように、アプリケーション出力を設定できます。結果が Amazon Redshift テーブルに書き込まれた後は、テーブルにクエリして `Ticker_Symbol` によってカウントグループの総数をコンピューティングできます。`XYZ` の場合、レコードが遅延したとしても総数は正確 (6\$11) です。 # 連続クエリ ストリーム上のクエリは、ストリーミングデータに対して連続して実行されます。この連続実行によって、アプリケーションが連続してストリーミングにクエリしアラートを生成する機能などのシナリオが可能になります。 「使用開始」の実習では、`SOURCE_SQL_STREAM_001` という名前のアプリケーション内ストリームを使用します。これはデモストリーム (Kinesis データストリーム) から連続して株価を受信します。スキーマは次のとおりです。 ``` (TICKER_SYMBOL VARCHAR(4), SECTOR varchar(16), CHANGE REAL, PRICE REAL) ``` 15 パーセントを超える株価の変動に関心があるとします。アプリケーションコードで次のクエリを使用できます。このクエリは連続して実行され、15 パーセントを超える株価の変動が検出された場合にレコードを発行します。 ``` SELECT STREAM TICKER_SYMBOL, PRICE FROM "SOURCE_SQL_STREAM_001" WHERE (ABS((CHANGE / (PRICE-CHANGE)) * 100)) > 15 ``` 次の手順を使用して Amazon Kinesis Data Analytics アプリケーションをセットアップし、このクエリをテストします。 **クエリをテストするには** 1. [「使用開始」実習](https://docs.aws.amazon.com/kinesisanalytics/latest/dev/get-started-exercise.html)に従ってアプリケーションを作成します。 1. アプリケーションコード内の `SELECT` ステートメントを前述の `SELECT` クエリに置き換えます。アプリケーションコードは次のようになります。 ``` CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" (ticker_symbol VARCHAR(4), price DOUBLE); -- CREATE OR REPLACE PUMP to insert into output CREATE OR REPLACE PUMP "STREAM_PUMP" AS INSERT INTO "DESTINATION_SQL_STREAM" SELECT STREAM TICKER_SYMBOL, PRICE FROM "SOURCE_SQL_STREAM_001" WHERE (ABS((CHANGE / (PRICE-CHANGE)) * 100)) > 15; ``` # ウィンドウクエリ アプリケーションコードの SQL クエリはアプリケーション内ストリームに対して連続で実行されます。アプリケーション内ストリームとは、アプリケーション内を常時流れる未バインドのデータのことです。したがって、常時更新されているこの入力から結果セットを得るために、時間と行の条件で定義されるウィンドウを使用してクエリをバインドする場合が多くあります。これらは*ウィンドウ SQL* とも呼ばれます。 時間ベースのウィンドウクエリの場合は、ウィンドウのサイズを時間で (たとえば、1 分のウィンドウ) 指定します。これには、一定間隔で増加するアプリケーション内ストリームにタイムスタンプ列が必要です。(新しい行のタイムスタンプが前の行と同じまたは前の行より大きい)。Amazon Kinesis Data Analytics は、各アプリケーション内ストリームに `ROWTIME` というタイムスタンプ列を提供します。時間ベースのクエリを指定するとき、この列を使用できます。アプリケーションで、他のタイムスタンプオプションを選択する場合もあります。詳細については、「[タイムスタンプと ROWTIME 列](timestamps-rowtime-concepts.md)」を参照してください。 行ベースのウィンドウクエリの場合は、列数の条件でウィンドウサイズを指定します。 アプリケーションの必要に応じて、タンブリングウィンドウ、スライディングウィンドウ、またはずらしウィンドウ方式でレコードを処理するクエリを指定できます。Kinesis Data Analytics では、次のウィンドウタイプがサポートされています。 + [Stagger Windows](stagger-window-concepts.md): データが届くと開く、キー付けされた時間ベースのウィンドウを使用してデータを集計するクエリ。キーによって、複数の重なり合うウィンドウが可能になります。タンブリングウィンドウと比較すると、Stagger Windows は遅延データまたは順序通りでないデータを削減するため、これは、時間ベースのウィンドウを使用してデータを集約する方法として推奨されます。 + [タンブリングウィンドウ](tumbling-window-concepts.md): 定期的に開閉する、個別の時間ベースのウィンドウを使用してデータを集計するクエリ。 + [スライディングウィンドウ](sliding-window-concepts.md): 固定時間または rowcount 間隔を使用して、データを継続的に集計するクエリ。 # Stagger Windows *ずらしウィンドウ*は、一貫性のない時間に届くデータのグループを分析するのに適したウィンドウ処理メソッドです。これは、関連する一連のセールスやログレコードなど、時系列分析のユースケースに適しています。 たとえば、[VPC フローログ](https://docs.aws.amazon.com/vpc/latest/userguide/flow-logs.html#flow-logs-limitations)には約 10 分のキャプチャウィンドウがあります。しかし、クライアントにデータを集約する場合は最大 15 分のキャプチャウィンドウを持つことができます。ずらしウィンドウは、これらのログを分析のために集計するのに理想的です。 ずらしウィンドウでは、タンブリングウィンドウが使用されたときなど、同じ時間制限付きウィンドウに収まらない関連レコードの問題を解決します。 ## Tumbling Windows の部分的な結果 遅延データまたは順序通りでないデータの集約に [タンブリングウィンドウ](tumbling-window-concepts.md) を使用する場合、一定の制限があります。 時間関連のデータのグループを分析するためにタンブリングウィンドウを使用する場合、個々のレコードは別々のウィンドウに分類される可能性があります。したがって、各ウィンドウの部分的な結果を後で組み合わせて、各レコードグループの完全な結果を得る必要があります。 次のタンブリングウィンドウクエリでは、レコードは行時間、イベント時間、およびティッカーシンボルによってウィンドウにグループ化されます。 ``` 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 分のタンブリングウィンドウに収まるわけではありません。 ![\[Tumbling windows diagram showing data grouping by row time, event time, and ticker symbol over two minutes.\]](http://docs.aws.amazon.com/ja_jp/kinesisanalytics/latest/dev/images/stagger_0.png) 前述の図には、以下のイベントが含まれます。 **** | ROWTIME | EVENT\$1TIME | TICKER\$1SYMBOL | | --- | --- | --- | | 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\$1TIME | TICKER\$1SYMBOL | 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); ``` 次の図では、イベントはイベント時間、およびティッカーシンボルによってずらしウィンドウに集計されます。 ![\[Diagram showing event aggregation into stagger windows by event time and ticker symbol.\]](http://docs.aws.amazon.com/ja_jp/kinesisanalytics/latest/dev/images/stagger_1.png) 前述の図には、タンブリングウィンドウアプリケーションが分析したものと同じイベントである、以下のイベントが含まれています。 **** | ROWTIME | EVENT\$1TIME | TICKER\$1SYMBOL | | --- | --- | --- | | 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\$1TIME | TICKER\$1SYMBOL | カウント | | --- | --- | --- | --- | | 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 WHERE <... optional statements...> WINDOWED BY STAGGER( PARTITION BY RANGE INTERVAL ); ``` # タンブリングウィンドウ (GROUP BY を使用した集計) ウィンドウクエリが各ウィンドウを重複しない方式で処理する場合、ウィンドウは *タンブリングウィンドウ*と呼ばれます。この場合、アプリケーション内ストリームの各レコードは特定のウィンドウに属します。これは 1 回 (そのレコードが属するウィンドウをクエリが処理するとき) のみ処理されます。 ![\[Timeline showing non-overlapping windows processing data streams at distinct time intervals.\]](http://docs.aws.amazon.com/ja_jp/kinesisanalytics/latest/dev/images/window-tumbling-20.png) たとえば、`GROUP BY` 句を使用した集計クエリは、タンブリングウィンドウの行を処理します。[「使用開始」実習](https://docs.aws.amazon.com/kinesisanalytics/latest/dev/get-started-exercise.html)のデモストリームは、アプリケーションのアプリケーション内ストリーム `SOURCE_SQL_STREAM_001` にマッピングされた株価データを受信します。このストリームには、次のスキーマがあります。 ``` (TICKER_SYMBOL VARCHAR(4), SECTOR varchar(16), CHANGE REAL, PRICE REAL) ``` アプリケーションコードで、1 分のウィンドウに対して各ティッカーでの合計 (最低、最高) 価格を検索するとします。以下のクエリを使用できます。 ``` SELECT STREAM ROWTIME, Ticker_Symbol, MIN(Price) AS Price, MAX(Price) AS Price FROM "SOURCE_SQL_STREAM_001" GROUP BY Ticker_Symbol, STEP("SOURCE_SQL_STREAM_001".ROWTIME BY INTERVAL '60' SECOND); ``` 上記は、時間ベースのウィンドウクエリの例です。クエリは、`ROWTIME` 値でレコードをグループ化します。分単位でレポートするために、`STEP` 関数は `ROWTIME` 値を直近の分に四捨五入します。 **注記** また、`FLOOR` 関数を使用してレコードをウィンドウにグループ化することもできます。ただし、`FLOOR` は時間値を時間単位 (時間、分、秒など) に丸めることのみできます。`STEP` は、値を任意の間隔 (たとえば 30 秒など) に丸めることができるため、レコードをタンブリングウィンドウにグループ化する場合に使用することをお勧めします。 このクエリは、重複しない (タンブリング) ウィンドウの例です。`GROUP BY` 句によって、レコードが 1 分のウィンドウにグループ化されます。各レコードは特定のウィンドウに属します (重複しない)。クエリでは、1 分ごとに 1 つの出力レコードが発行され、特定の分にレコードされた最低/最高ティッカー価格が提供されます。このタイプのクエリは、入力データストリームから定期的にレポートを生成する場合に便利です。この例では、1 分ごとにレポートが生成されます。 **クエリをテストするには** 1. [「使用開始」実習](https://docs.aws.amazon.com/kinesisanalytics/latest/dev/get-started-exercise.html)に従ってアプリケーションをセットアップします。 1. アプリケーションコード内の `SELECT` ステートメントを前述の `SELECT` クエリに置き換えます。アプリケーションコードは次のようになります。 ``` CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" ( ticker_symbol VARCHAR(4), Min_Price DOUBLE, Max_Price DOUBLE); -- CREATE OR REPLACE PUMP to insert into output CREATE OR REPLACE PUMP "STREAM_PUMP" AS INSERT INTO "DESTINATION_SQL_STREAM" SELECT STREAM Ticker_Symbol, MIN(Price) AS Min_Price, MAX(Price) AS Max_Price FROM "SOURCE_SQL_STREAM_001" GROUP BY Ticker_Symbol, STEP("SOURCE_SQL_STREAM_001".ROWTIME BY INTERVAL '60' SECOND); ``` # スライディングウィンドウ `GROUP BY` を使用してレコードをグループ化する代わりに、時間ベースまたは行ベースのウィンドウを定義できます。そのためには、`WINDOW` 句を明示的に追加します。 この場合、ウィンドウが時間と共にスライドしながら、新しいレコードがストリームに現れると Amazon Kinesis Data Analytics が出力を発行します。Kinesis Data Analytics は、ウィンドウの行を処理して出力を発行します。このタイプの処理ではウィンドウが重複するだけでなく、レコードが複数のウィンドウの一部となり、ウィンドウごとに処理される場合があります。次の例では、スライディングウィンドウについて説明します。 ストリームのレコードをカウントする簡単なクエリを考えます。この例では、5 秒のウィンドウを前提としています。次のストリームの例では、新しいレコードが t1、t2、t6、t7 の時間に受信され、t8 秒では同時に 3 つのレコードを受信しています。 ![\[Timeline showing record arrivals at t1, t2, t6, t7, and multiple at t8 within a 5-second window.\]](http://docs.aws.amazon.com/ja_jp/kinesisanalytics/latest/dev/images/sliding-10.png) 以下に留意してください。 + この例では、5 秒のウィンドウを前提としています。5 秒ウィンドウは時間とともに継続的にスライドします。 + 行がウィンドウに入力されるごとに、出力行がスライディングウィンドウによって発行されます。アプリケーションを起動してすぐは、まだ 5 秒のウィンドウが経過していなくても、ストリームで受信された新しいレコードのそれぞれに対してクエリが出力を発行します。たとえば、1 秒目と 2 秒目にレコードが現れると、クエリは出力を発行します。その後、クエリは 5 秒ウィンドウでレコードを処理します。 + ウィンドウは時間とともにスライドします。古いレコードがウィンドウから押し出されても、その 5 秒ウィンドウに含まれるストリームに新しいレコードがない限り、クエリは出力を発行しません。 クエリが t0 に実行を開始するとします。そして次のようになります。 1. t0 にクエリが開始されます。この時点ではレコードがないため、クエリは出力 (カウント値) を発行しません。 ![\[Timeline showing a stream starting at t0 with no output initially indicated.\]](http://docs.aws.amazon.com/ja_jp/kinesisanalytics/latest/dev/images/sliding-t0.png) 1. 時間 t1 に、新しいレコードがストリームに現れ、クエリはカウント値 1 を発行します。 ![\[Timeline showing a stream with a record appearing at time t1, and an arrow pointing to t0.\]](http://docs.aws.amazon.com/ja_jp/kinesisanalytics/latest/dev/images/sliding-t1.png) 1. 時間 t2 に、別のレコードが現れ、クエリはカウント値 2 を発行します。 ![\[Timeline showing stream events at different time points, with two vertical bars at the end.\]](http://docs.aws.amazon.com/ja_jp/kinesisanalytics/latest/dev/images/sliding-t2.png) 1. 5 秒ウィンドウは時間とともにスライドします。 + t3 では、スライディングウィンドウは t3 から t0 です。 + t4 (スライディングウィンドウは t4 から t0) + t5 では、スライディングウィンドウは t5 から t0 です。 この間、5 秒ウィンドウのレコードはまったく同じです。新規レコードはありません。そのため、クエリは出力を発行しません。 ![\[Timeline showing stream with multiple time points and colored rectangles representing data windows.\]](http://docs.aws.amazon.com/ja_jp/kinesisanalytics/latest/dev/images/sliding-t3-4-5.png) 1. t6 時、5 秒ウィンドウは (t6 から t1) です。クエリは、t6 で新しいレコードを検出するため、出力 2 を発行します。t1 のレコードはウィンドウ内になくなったため、カウントされません。 ![\[Timeline showing stream events at different time points with a sliding 5-second window.\]](http://docs.aws.amazon.com/ja_jp/kinesisanalytics/latest/dev/images/sliding-t6.png) 1. t7 時、5 秒ウィンドウは (t7 から t2) です。クエリは、t7 で新しいレコードを検出するため、出力 2 を発行します。t2 のレコードは 5 秒ウィンドウ内になくなったため、カウントされません。 ![\[Timeline showing stream events and time points from t0 to t7, with a 5-second window highlighted.\]](http://docs.aws.amazon.com/ja_jp/kinesisanalytics/latest/dev/images/sliding-t7.png) 1. t8 時、5 秒ウィンドウは (t8 から t3) です。クエリが 3 つの新しいレコードを検出したため、レコードカウント 5 を発行します。 ![\[Timeline showing stream events with orange bars representing record counts at different time intervals.\]](http://docs.aws.amazon.com/ja_jp/kinesisanalytics/latest/dev/images/sliding-t8.png) 要約すると、このウィンドウは固定サイズであり、時間とともにスライドします。クエリは新しいレコードが現れたときに出力を発行します。 **注記** スライディングウィンドウの使用は1 時間以内にすることをお勧めします。これよりも長いウィンドウを使用する場合、通常のシステムメンテナンス後のアプリケーションの再起動に時間がかかります。これは、ソースデータを再度ストリームから読み取る必要があるためです。 以下は、`WINDOW` 句を使用してウィンドウを定義し集計を実行するクエリの例です。クエリが `GROUP BY` を指定しないため、このクエリではスライディングウィンドウの方法を使用してストリームのレコードを処理します。 ## 例 1: 1 分のスライディングウィンドウを使用してストリームを処理する アプリケーション内ストリームに入力する「はじめに」実習のデモストリーム、`SOURCE_SQL_STREAM_001` を考えてみます。スキーマは次のとおりです。 ``` (TICKER_SYMBOL VARCHAR(4), SECTOR varchar(16), CHANGE REAL, PRICE REAL) ``` 1 分のスライディングウィンドウを使用して、アプリケーションで集計をコンピューティングすると仮定します。つまり、ストリームに現れる新しいレコードそれぞれについて、前の 1 分ウィンドウのレコードの集計を適用することで、アプリケーションに出力を発行させます。 以下の時間ベースのウィンドウクエリを使用できます。クエリは、`WINDOW` 句を使用して 1 分間隔の範囲を定義します。`WINDOW` 句の `PARTITION BY` はスライディングウィンドウ内のティッカー値でレコードをグループ化します。 ``` SELECT STREAM ticker_symbol, MIN(Price) OVER W1 AS Min_Price, MAX(Price) OVER W1 AS Max_Price, AVG(Price) OVER W1 AS Avg_Price FROM "SOURCE_SQL_STREAM_001" WINDOW W1 AS ( PARTITION BY ticker_symbol RANGE INTERVAL '1' MINUTE PRECEDING); ``` **クエリをテストするには** 1. [「使用開始」実習](https://docs.aws.amazon.com/kinesisanalytics/latest/dev/get-started-exercise.html)に従ってアプリケーションをセットアップします。 1. アプリケーションコード内の `SELECT` ステートメントを前述の `SELECT` クエリに置き換えます。アプリケーションコードは次のようになります。 ``` CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" ( ticker_symbol VARCHAR(10), Min_Price double, Max_Price double, Avg_Price double); CREATE OR REPLACE PUMP "STREAM_PUMP" AS INSERT INTO "DESTINATION_SQL_STREAM" SELECT STREAM ticker_symbol, MIN(Price) OVER W1 AS Min_Price, MAX(Price) OVER W1 AS Max_Price, AVG(Price) OVER W1 AS Avg_Price FROM "SOURCE_SQL_STREAM_001" WINDOW W1 AS ( PARTITION BY ticker_symbol RANGE INTERVAL '1' MINUTE PRECEDING); ``` ## 例 2: スライディングウインドウに集計を適用するクエリ デモストリームに対する次のクエリは、10 秒ウィンドウの各ティッカーの価格の変動パーセントの平均を返します。 ``` SELECT STREAM Ticker_Symbol, AVG(Change / (Price - Change)) over W1 as Avg_Percent_Change FROM "SOURCE_SQL_STREAM_001" WINDOW W1 AS ( PARTITION BY ticker_symbol RANGE INTERVAL '10' SECOND PRECEDING); ``` **クエリをテストするには** 1. [「使用開始」実習](https://docs.aws.amazon.com/kinesisanalytics/latest/dev/get-started-exercise.html)に従ってアプリケーションをセットアップします。 1. アプリケーションコード内の `SELECT` ステートメントを前述の `SELECT` クエリに置き換えます。アプリケーションコードは次のようになります。 ``` CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" ( ticker_symbol VARCHAR(10), Avg_Percent_Change double); CREATE OR REPLACE PUMP "STREAM_PUMP" AS INSERT INTO "DESTINATION_SQL_STREAM" SELECT STREAM Ticker_Symbol, AVG(Change / (Price - Change)) over W1 as Avg_Percent_Change FROM "SOURCE_SQL_STREAM_001" WINDOW W1 AS ( PARTITION BY ticker_symbol RANGE INTERVAL '10' SECOND PRECEDING); ``` ## 例 3: 同じストリームの複数のスライディングウィンドウからのデータのクエリ 同じストリームに対して定義された別々のスライディングウィンドウを使用して各列値を計算し出力を発行するクエリを作成できます。 次の例では、クエリは出力ティッカー、価格、a2、a10 を発行します。また、2 行の移動平均に 10 行の移動平均を交えたティッカーシンボルについて出力を発行します。列 `a2` および `a10` の値は、2 行および 10 行のスライディングウィンドウから取得されます。 ``` CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" ( ticker_symbol VARCHAR(12), price double, average_last2rows double, average_last10rows double); CREATE OR REPLACE PUMP "myPump" AS INSERT INTO "DESTINATION_SQL_STREAM" SELECT STREAM ticker_symbol, price, avg(price) over last2rows, avg(price) over last10rows FROM SOURCE_SQL_STREAM_001 WINDOW last2rows AS (PARTITION BY ticker_symbol ROWS 2 PRECEDING), last10rows AS (PARTITION BY ticker_symbol ROWS 10 PRECEDING); ``` デモストリームに対してこのクエリをテストするには、「[例 1](#sliding-ex1)」で説明されているテスト手順に従います。 # ストリーミングデータオペレーション: ストリーム結合 アプリケーションに複数のアプリケーション内ストリームを指定できます。これらストリームに届くデータを関連付ける `JOIN` クエリを記述できます。たとえば、以下のアプリケーション内ストリームがあるとします。 + **OrderStream** – 発注された株注文を受け取ります。 ``` (orderId SqlType, ticker SqlType, amount SqlType, ROWTIME TimeStamp) ``` + **TradeStream** – それらの注文に対する株取引結果を受け取ります。 ``` (tradeId SqlType, orderId SqlType, ticker SqlType, amount SqlType, ticker SqlType, amount SqlType, ROWTIME TimeStamp) ``` 以下は、これらのストリームのデータを関連付ける `JOIN` クエリの例です。 ## 例 1: 注文が出されてから 1 分以内に取引があった注文をレポートする この例では、クエリは `OrderStream` と `TradeStream` の両方を結合します。ただし、注文から 1 分で発生した取引のみが必要であるため、クエリで `TradeStream` に対して 1 分ウィンドウを定義します。ウィンドウクエリについては、「[スライディングウィンドウ](sliding-window-concepts.md)」を参照してください。 ``` SELECT STREAM ROWTIME, o.orderId, o.ticker, o.amount AS orderAmount, t.amount AS tradeAmount FROM OrderStream AS o JOIN TradeStream OVER (RANGE INTERVAL '1' MINUTE PRECEDING) AS t ON o.orderId = t.orderId; ``` `WINDOW` 句を使用してウィンドウを明示的に定義し、前述のクエリを次のように記述できます。 ``` SELECT STREAM ROWTIME, o.orderId, o.ticker, o.amount AS orderAmount, t.amount AS tradeAmount FROM OrderStream AS o JOIN TradeStream OVER t ON o.orderId = t.orderId WINDOW t AS (RANGE INTERVAL '1' MINUTE PRECEDING) ``` このクエリをアプリケーションコードに含めると、アプリケーションコードは連続実行されます。`OrderStream` の各到着レコードについて、注文の発注に続いて 1 分ウィンドウ内で取引があれば、アプリケーションで出力が発行されます。 前述のクエリでの結合は内部結合であり、クエリは `TradeStream` に一致するレコードがある `OrderStream` のレコードを発行します (逆も同様です)。外部結合を使用すると、別の興味深いシナリオを作成できます。株注文が発注されてから 1 分以内に取引がない株注文と、同じウィンドウでレポートされた別の注文に対する取引を指定するとします。これは、*外部結合*の例です。 ``` SELECT STREAM ROWTIME, o.orderId, o.ticker, o.amount AS orderAmount, t.ticker, t.tradeId, t.amount AS tradeAmount, FROM OrderStream AS o LEFT OUTER JOIN TradeStream OVER (RANGE INTERVAL '1' MINUTE PRECEDING) AS t ON o.orderId = t.orderId; ```