ステップ 2: アプリケーションの作成

このセクションでは、次のように Amazon Kinesis Data Analytics アプリケーションを作成します。

• ステップ 1: 準備 で作成した Kinesis データストリームをストリーミングソースとして使用するように、アプリケーション入力を設定します。

• コンソールで [異常検出] テンプレートを使用します。

アプリケーションを作成するには

1. Kinesis Data Analytics の開始方法の実習のステップ 1、2、および 3 (「ステップ 3.1: アプリケーションの作成」を参照) に従います。

   • ソース設定で、以下を実行します。

     • 前のセクションで作成したストリーミングソースを指定します。

     • コンソールがスキーマを推測した後、スキーマを編集し、heartRate 列の型を INTEGER に設定します。

       心拍値のほとんどは正常で、検出プロセスはこの列に TINYINT 型を割り当てることがほとんどです。ただし、低い割合で値が高い心拍数を示します。これらの高い値が TINYINT 型に合わない場合、Kinesis Data Analytics はそれらの列をエラーストリームに送ります。データ型を INTEGER に更新して、生成された心拍数データのすべてに対応できるようにします。

   • コンソールで [異常検出] テンプレートを使用します。次にテンプレートコードを更新して適切な列名を指定します。

2. 列名を指定してアプリケーションコードを更新します。その結果アプリケーションコードは次のようになります (このコードを SQL エディタに貼り付けます)。

   --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;

3. SQL コードを実行して Kinesis Data Analytics コンソールで結果を確認します。

   

次のステップ

ステップ 3: アプリケーション出力を設定する