例: イベントのタイムスタンプを使用したタンブリングウィンドウ

AWS Management Consoleにサインインして、Kinesis コンソール (https://console.aws.amazon.com/kinesis) を開きます。

ナビゲーションペインで、[データストリーム] を選択します。

[Kinesis ストリームの作成] を選択後、1 つのシャードがあるストリームを作成します。詳細については、「Amazon Kinesis Data Streams デベロッパーガイド」の「Create a Stream」を参照してください。

本稼働環境の Kinesis データストリームにレコードを書き込むには、Kinesis Client Library または Kinesis Data Streams API を使用することをお勧めします。わかりやすいように、この例では、以下の Python スクリプトを使用してレコードを生成します。サンプルのティッカーレコードを入力するには、このコードを実行します。このシンプルなコードによって、ランダムなティッカーレコードが連続してストリームに書き込まれます。後のステップでアプリケーションスキーマを生成できるように、スクリプトを実行したままにしておきます。

 
import datetime
import json
import random
import boto3

STREAM_NAME = "ExampleInputStream"


def get_data():
    return {
        "EVENT_TIME": datetime.datetime.now().isoformat(),
        "TICKER": random.choice(["AAPL", "AMZN", "MSFT", "INTC", "TBV"]),
        "PRICE": round(random.random() * 100, 2),
    }


def generate(stream_name, kinesis_client):
    while True:
        data = get_data()
        print(data)
        kinesis_client.put_record(
            StreamName=stream_name, Data=json.dumps(data), PartitionKey="partitionkey"
        )


if __name__ == "__main__":
    generate(STREAM_NAME, boto3.client("kinesis"))

https://console.aws.amazon.com/kinesisanalytics にある Managed Service for Apache Flink コンソールを開きます。

[アプリケーションの作成] を選択し、アプリケーション名を入力して、[アプリケーションの作成] を選択します。

アプリケーション詳細ページで、[ストリーミングデータの接続] を選択してソースに接続します。

[ソースに接続] ページで、以下の操作を実行します。

1. 前のセクションで作成したストリームを選択します。

2. [スキーマの検出] を選択します。作成されたアプリケーション内ストリーム用の推測スキーマと、推測に使用されたサンプルレコードがコンソールに表示されるまで待ちます。推測されたスキーマには 3 つの列があります。

3. [スキーマの編集] を選択します。[EVENT_TIME] 列の [列のタイプ] を TIMESTAMP に変更します。

4. [Save schema and update stream samples] を選択します。コンソールでスキーマが保存されたら、[終了] を選択します。

5. [Save and continue] を選択します。

アプリケーション詳細ページで、[SQL エディタに移動] を選択します。アプリケーションを起動するには、表示されたダイアログボックスで [はい、アプリケーションを起動します] を選択します。

SQL エディタで、次のように、アプリケーションコードを作成してその結果を確認します。

1. 次のアプリケーションコードをコピーしてエディタに貼り付けます。

   CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" (EVENT_TIME timestamp, TICKER VARCHAR(4), min_price REAL, max_price REAL);
   
   CREATE OR REPLACE PUMP "STREAM_PUMP" AS 
     INSERT INTO "DESTINATION_SQL_STREAM" 
       SELECT STREAM STEP("SOURCE_SQL_STREAM_001".EVENT_TIME BY INTERVAL '60' SECOND),
           TICKER,
            MIN(PRICE) AS MIN_PRICE,
            MAX(PRICE) AS MAX_PRICE
       FROM    "SOURCE_SQL_STREAM_001"
       GROUP BY TICKER, 
                STEP("SOURCE_SQL_STREAM_001".ROWTIME BY INTERVAL '60' SECOND), 
                STEP("SOURCE_SQL_STREAM_001".EVENT_TIME BY INTERVAL '60' SECOND);

2. [Save and run SQL] を選択します。

   [リアルタイム分析] タブに、アプリケーションで作成されたすべてのアプリケーション内ストリームが表示され、データを検証できます。