Kinesis Data AnalyticsでRandom Cut Forest（異常検知）を試してみた

Kinesis Data AnalyticsのSQLアプリケーションでRandom Cut Forestを試してみました

#Kinesis Data Analytics

#Python

#Kinesis Data Streams

おざわ（じ）

2022.10.14

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

こんにちは！ブログ筋を鍛えたい入社12日目のおざわ（じ）です。

AWS学習とブログネタ探しも兼ねて、10月8日（土）に開催されたJAWS DAYS 2022に参加してきました。

私はオンライン参加でしたが、日本各地にサテライト会場も設置されており、各会場の様子もバーチャルで覗くことができました。セッションの合間にはノリノリのBGMで実行委員の皆さんの紹介もあり、オンラインでありながら会場にいるかのように感じられる素晴らしいイベントでした。

概要

さて、本記事の内容ですが、今回のイベントで個人的に興味を惹かれたソラコム株式会社松下さんの発表内容を手元の環境で試してみました。

使用したサービスは「Kinesis Data Streams」と「Kinesis Data Analytics」の2つになります。異常検知に関しては、発表と同じKinesis Data Analyticsに用意されているRandom Cut Forest（RCF）というアルゴリズムを使用します。

今回はAWSコンソールで異常検知できたことを確かめるのみでしたが、異常検知した後に何らかのアクションを行う場合は、Lambdaを使ったこちらの記事もご参照ください。

やったこと

Kinesis Data Streamの設定
スクリプトの準備
Kinesis Data Analyticsの設定
正常値データを投入
異常値を投入

1. Kinesis Data Streamの設定

まずはデータ入力先のデータストリームをCLIで作成します。

aws kinesis create-stream \
--stream-name VolumeStream \
--shard-count 1 \
--region ap-northeast-1

2. スクリプトの準備

今回はストリーム用のデータとして、マイクからの音量をリアルタイムに流し込んでみようと思います。python-sounddeviceが手軽に使えそうだったので、今回はこちらを使用しました。

import json
import datetime

import boto3
import sounddevice as sd
import numpy as np

STREAM_NAME = "VolumeStream"
DURATION = 15

kinesis = boto3.client("kinesis")

def put_volume(indata, outdata, frames, time, status):
    volume = np.linalg.norm(indata) * DURATION
    data = {
            'event_time': datetime.datetime.now().isoformat(),
            'volume': volume
    }
    print (data)

    kinesis.put_record(
            StreamName=STREAM_NAME,
            Data=json.dumps(data),
            PartitionKey="123")

with sd.Stream(samplerate=44100, callback=put_volume):
    sd.sleep(int(1000 * DURATION))

次のステップで「スキーマの検出」を行うので、先にデータを入れておきます。

> python sound.py
{'event_time': '2022-10-11T16:13:02.150653', 'volume': 0.0}
{'event_time': '2022-10-11T16:13:07.619224', 'volume': 0.19032749813050032}
{'event_time': '2022-10-11T16:13:07.660909', 'volume': 1.829547956585884}
{'event_time': '2022-10-11T16:13:07.700621', 'volume': 1.3599266856908798}

3. Kinesis Data Analyticsの設定

SQLアプリケーションを作成します。

新しいプロジェクトでは、SQLアプリケーションよりも新しいKinesis Data Analytics Studioを使用することが推奨されていますが、今回は練習も兼ねて以前のインターフェースからやってみます。 Kinesis Data Analytics Studioの利用方法に関しては、参考記事にリンクを載せていますので、そちらをご確認いただければと思います。

アプリケーションが作成できたら、ソースとなるストリームに先ほど設定したものを割り当てます。参照ボタンからストリームを選択して「スキーマを検出」ボタンをクリックするとスキーマが検出されます。

「変更を保存」をクリックしたら、次に「実行」をクリックします

SQLコードの設定

「リアルタイム分析」のタブをクリックして「設定」ボタンを押すと、下のような画面が表示されますので、ここからSQLコードの設定を行います。

--Creates a temporary stream.
CREATE OR REPLACE STREAM "TEMP_STREAM" (
	        "volume"        INTEGER,
	        "ANOMALY_SCORE"    DOUBLE);

--Creates another stream for application output.	        
CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" (
	        "volume"        INTEGER,
	        "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 "volume",
        ANOMALY_SCORE 
      FROM TABLE(
        RANDOM_CUT_FOREST(
           --inputStream
           CURSOR(SELECT STREAM * FROM "SOURCE_SQL_STREAM_001"),
           --numberOfTrees
           100,
           --subSampleSize
           10,
           --timeDecay
           100,
           --shingleSize
           1
        ));

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

SQLを入力したら「アプリケーションを保存して実行」をクリックします。

4. データを流す

マイクを挿してからスクリプトを実行してデータを流し込んでみます。

> python sound.py
{'event_time': '2022-10-11T16:16:04.614177', 'volume': 1.556536853313446}
{'event_time': '2022-10-11T16:16:04.655622', 'volume': 0.9072657488286495}
{'event_time': '2022-10-11T16:16:04.693902', 'volume': 0.9062547236680984}
{'event_time': '2022-10-11T16:16:04.732057', 'volume': 0.7885556481778622}

最初はなにも表示されていませんので自然音のみで放置してみます。