例 : ストリーム上のホットスポットの検出 (HOTSPOTS 関数)

ストリーミングソースのセットアップ – Kinesis ストリームをセットアップして、次のようにサンプル座標データを書き込みます。

{"x": 7.921782426109737, "y": 8.746265312709893, "is_hot": "N"}
{"x": 0.722248626528026, "y": 4.648868803193405, "is_hot": "Y"}

この例では、ストリームに入力するための Python スクリプトを提供しています。x および y 値はランダムに生成され、一部のレコードは特定の場所の周りにクラスター化されます。

スクリプトがホットスポットの一部として意図的に値を生成した場合、is_hot フィールドはインジケータとして提供されます。これは、ホットスポット検出関数が正常に動作しているかどうかを評価するのに役立ちます。

アプリケーションの作成 – AWS Management Console を使用して Kinesis Data Analytics アプリケーションを作成します。ストリーミングソースをアプリケーション内ストリーム (SOURCE_SQL_STREAM_001) にマッピングして、アプリケーション入力を設定します。アプリケーションが起動すると、Kinesis Data Analytics は継続的にストリーミングソースを読み取り、アプリケーション内ストリームにレコードを挿入します。

この演習では、アプリケーションに次のコードを使用します。

CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" (
    "x" DOUBLE, 
    "y" DOUBLE, 
    "is_hot" VARCHAR(4),
    HOTSPOTS_RESULT VARCHAR(10000)
); 
CREATE OR REPLACE PUMP "STREAM_PUMP" AS 
    INSERT INTO "DESTINATION_SQL_STREAM" 
    SELECT "x", "y", "is_hot", "HOTSPOTS_RESULT" 
    FROM TABLE (
        HOTSPOTS(   
            CURSOR(SELECT STREAM "x", "y", "is_hot" FROM "SOURCE_SQL_STREAM_001"), 
            1000, 
            0.2, 
            17)
    );

コードは SOURCE_SQL_STREAM_001 の行を読み取り、重要なホットスポットを分析し、結果のデータを別のアプリケーション内ストリーム (DESTINATION_SQL_STREAM) に書き込みます。ポンプを使用して、アプリケーション内ストリームに行を挿入します。詳細については、「アプリケーション内ストリームとポンプ」を参照してください。

出力の設定 – アプリケーションから別の Kinesis データストリームである外部送信先にデータを送信するように、アプリケーション出力を設定します。ホットスポットのスコアを確認し、ホットスポットが発生したことを示すスコア (およびアラートが必要なスコア) を判断します。AWS Lambda 関数を使用して、ホットスポット情報をさらに処理し、アラートを設定することができます。

出力を確認する – この例には、出力ストリームからデータを読み込んでグラフィカルに表示する JavaScript アプリケーションが含まれているので、アプリケーションが生成するホットスポットをリアルタイムで表示できます。