例: ストリームでデータの異常を検出する (RANDOM_CUT_FOREST 関数) - Amazon Kinesis Data Analytics for SQL Applications 開発者ガイド

新しいプロジェクトでは、SQL アプリケーション用 Kinesis Data Analytics ではなく、新しい Kinesis Data Analytics スタジオを使用することをお勧めします。Kinesis Data Analytics Studio は使いやすさと高度な分析機能を兼ね備えているため、高度なストリーム処理アプリケーションを数分で構築できます。

翻訳は機械翻訳により提供されています。提供された翻訳内容と英語版の間で齟齬、不一致または矛盾がある場合、英語版が優先します。

例: ストリームでデータの異常を検出する (RANDOM_CUT_FOREST 関数)

Amazon Kinesis Data Analytics には、数値列の値に基づいて各レコードに異常スコアを割り当てることができる関数 (RANDOM_CUT_FOREST) が用意されています。詳細については、Amazon Kinesis Data Analytics SQL リファレンスのRANDOM_CUT_FOREST関数」を参照してください。

この実習では、アプリケーションのストリーミングソースのレコードに異常スコアを割り当てるアプリケーションコードを作成します。アプリケーションをセットアップするには、以下を実行します。

  1. ストリーミングソースの設定 — Kinesis データストリームを設定し、heartRate次に示すようにサンプルデータを書き込みます。

    {"heartRate": 60, "rateType":"NORMAL"} ... {"heartRate": 180, "rateType":"HIGH"}

    この手順では、ストリームに入力するための Python スクリプトを提供しています。heartRate 値はランダムに生成されます。レコードの 99 パーセントは heartRate 値が 60 から 100 の間で、heartRate 値の 1 パーセントのみが 150 から 200 の間です。したがって、heartRate 値が 150 から 200 の間のレコードは異常です。

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

  3. アプリケーションコードの指定 – この例では、次のアプリケーションコードを使用します。

    --Creates a temporary stream. CREATE OR REPLACE STREAM "TEMP_STREAM" ( "heartRate" INTEGER, "rateType" varchar(20), "ANOMALY_SCORE" DOUBLE); --Creates another stream for application output. CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" ( "heartRate" INTEGER, "rateType" varchar(20), "ANOMALY_SCORE" DOUBLE); -- Compute an anomaly score for each record in the input stream -- using Random Cut Forest CREATE OR REPLACE PUMP "STREAM_PUMP" AS INSERT INTO "TEMP_STREAM" SELECT STREAM "heartRate", "rateType", ANOMALY_SCORE FROM TABLE(RANDOM_CUT_FOREST( CURSOR(SELECT STREAM * FROM "SOURCE_SQL_STREAM_001"))); -- Sort records by descending anomaly score, insert into output stream CREATE OR REPLACE PUMP "OUTPUT_PUMP" AS INSERT INTO "DESTINATION_SQL_STREAM" SELECT STREAM * FROM "TEMP_STREAM" ORDER BY FLOOR("TEMP_STREAM".ROWTIME TO SECOND), ANOMALY_SCORE DESC;

    コードは SOURCE_SQL_STREAM_001 の行を読み取り、異常スコアを割り当て、結果の行を別のアプリケーション内ストリーム (TEMP_STREAM) に書き込みます。次に、アプリケーションコードはTEMP_STREAM のレコードをソートして、結果を別のアプリケーション内ストリーム (DESTINATION_SQL_STREAM) に保存します。ポンプを使用して、アプリケーション内ストリームに行を挿入します。詳細については、「アプリケーション内ストリームとポンプ」を参照してください。

  4. 出力を設定 — データを、別の Kinesis データストリームである外部の宛先に保持するようにアプリケーション出力を設定します。DESTINATION_SQL_STREAM各レコードに割り当てられた異常スコアを確認して、どのスコアが異常を示しているか (また、アラートが必要か) を調べるのは、アプリケーションの範囲外です。このような異常スコアを処理してアラートを設定するには、AWS Lambda 関数を使用できます。

演習では、米国東部 (バージニア北部) (us-east-1) を使用してこれらのストリームとアプリケーションを作成します。他のリージョンも使用する場合は、それに応じてコードを更新する必要があります。

次のステップ

ステップ 1: 準備