例: ストリーム上のデータ異常の検出 (RANDOM_CUT_FOREST 関数)

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

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

1. ストリーミングソースを設定する— Kinesis Data Stream をセットアップして、サンプルを記述します。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. 出力の設定— のデータを永続化して、データ永続化してDESTINATION_SQL_STREAM外部宛先に書き込むと、別の Kinesis データストリームに書き込むことができます。各レコードに割り当てられた異常スコアを確認して、どのスコアが異常を示しているか (また、アラートが必要か) を調べるのは、アプリケーションの範囲外です。このような異常スコアを処理してアラートを設定するには、AWS Lambda 関数を使用できます。

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

トピック

• ステップ 1: Prepare
• ステップ 2: アプリケーションを作成します。
• ステップ 3: アプリケーション出力の設定
• ステップ 4: 出力を検証する

次のステップ

ステップ 1: Prepare