はじめに
Databricks の Notebook から Azure Anomaly Detector の API を叩いて結果を可視化するところまでを行います。
連絡目次
1. 導入/環境設定
2. Collaborative Notebook でデータ可視化
3. Anomaly Detector をデータ探索ツールとして使ってみる
→ 本稿
4. 1つ目のモデル構築 (データの偏り 未考慮)
5. 2つ目のモデル構築 (データの偏り 考慮)
Anomaly Detector とは
Anomaly Detector はマネージド型AIサービスとして提供されている Azure Cognitive Services の一つです。
- 時系列データセットから異常検出
- 教師なし学習
- 異常に対する感度を微調整可能
- APIでアプリケーションへ組み込み可能
- データ型や欠損値に対する制約がちょっと厳しい
予測結果の可視化一例
本連載の主題である異常検知モデルには利用していませんが、時系列データの特徴をさっと把握したいときに使えるかも、入力/出力ともにシンプルなので Detabricks の可視化機能と相性いいかも、ということで試してみました。
テストデータ生成
Anomaly Detector に投入する Body データは以下を満たす必要があります。
- UTC 形式のタイムスタンプ (カラム名 = timestamp)
- 数値 (異常検出対象。カラム名 = value)
今回は前記事で周期性を明確に見られた区間を抜粋し、 Step が前者、トランザクション数が後者になるようにデータを成型、テストデータとしました。
1# step から時系列のカラム作成 1/6-17 のみ抜粋。タイプはTRANSFER
2query1 = '''
3 CONCAT('2020-01-', LPAD(ceil(step / 24), 2, '0'), ' ', LPAD((step - 1) % 24, 2, '0'), ':00:00') as date_time_str
4 , count(1) as value
5 financial
6 type='TRANSFER' AND step < 409 AND step > 120
7GROUP BY step
8ORDER BY step
9'''
10
11agged_df = spark.sql(query1)
12agged_df.createOrReplaceTempView("agged_view")
13
14# タイムスタンプ作成
15query2 = '''
16 CAST(date_time_str as timestamp) as timestamp
17 , value
18 agged_view
19'''
20
21agged_df2 = spark.sql(query2)
22agged_df2.createOrReplaceTempView("agged_view2")
23
24# リクエスト用データフレーム作成
25agged_pd = agged_df2.select("*").toPandas()
26agged_pd.loc[:, ["timestamp"]] = agged_pd["timestamp"].dt.strftime("%Y-%m-%dT%H:%M:%SZ")テストデータを棒グラフにしてみます。数値の吹き出しがある箇所でトランザクションの少ないことはすぐ分かりますが、他にも特筆すべきポイントがあるかもしれません。
1SELECT * FROM agged_view2result
環境変数設定
まず Anomaly Detector のリソースを作成します。 (こちらの記事が参考になります)
まず Anomaly Detector のリソースを作成します。 (こちらの記事が参考になります)
エンドポイント URL と API Key を取得したら、クラスタの環境変数に追加しておきます。(Notebook に平打ちするのはセキュリティ上よろしくない)
関数定義
パラメータや実行関数を定義します。
1import pandas as pd
2import json
3import requests
4
5url = "/anomalydetector/v1.0/timeseries/entire/detect"
6endpoint = os.environ["ANOMALY_DETECTOR_ENDPOINT"]
7subscription_key = os.environ["ANOMALY_DETECTOR_KEY"]
8
9# 作成したデータフレームを json に変換
10series = json.loads(agged_pd.to_json(orient="records"))
11
12# 時系列データの粒度と異常値の感度を Bodyに入れる
13body = {
14 "series": series
15 ,"granularity": "hourly"
16 , "sensitivity": 50
17}
18
19def send_request(endpoint, url, subscription_key, request_data):
20 headers = {'Content-Type': 'application/json', 'Ocp-Apim-Subscription-Key': subscription_key}
21 response = requests.post(endpoint+url, data=json.dumps(request_data), headers=headers)
22 return json.loads(response.content.decode("utf-8"))
23
24def detect_batch(request_data):
25 print("Detecting anomalies as a batch")
26# リクエスト送信、返り値を出力
27 result = send_request(endpoint, url, subscription_key, request_data)
28 print(json.dumps(result, indent=4))
29 return result
30
31 if result.get('code') is not None:
32 print("Detection failed. ErrorCode:{}, ErrorMessage:{}".format(result['code'], result['message']))
33 else:
34# データセット内の異常値を表示
35 anomalies = result["isAnomaly"]
36 print("Anomalies detected in the following data positions:")
37
38 for x in range(len(anomalies)):
39 if anomalies[x]:
40 print (x, request_data['series'][x]['value'])実行・結果可視化
API にデータを投げ、元のデータフレームと結合します。
1results = detect_batch(body)
2mgd_df = pd.concat([agged_pd, pd.DataFrame(results)], axis=1)isAnomaly (予測結果。1=異常、0=通常) に絞ってグラフ化します。吹き出しの出ている箇所は目視では気が付きませんでしたが、確かに別の日の同じ時間に比べるとトランザクション数が低いように思われます。
1display(mgd_df)result
plot option
まとめ
私たちはグラフにするとなんとなく分かったような気になってしまう傾向があります。Anonaly Detector は先入観なしに時系列データを見てみたい時に便利かもしれませんね。次回は実際に決定木を用いてモデルを構築します。お楽しみに!
参考リンク
Anomaly Detector API の使用に関するベスト プラクティス
Cognitive Service: Anomaly Detector APIを試す
クイック スタート:Anomaly Detector REST API および Python を使用し、時系列データ内の異常を検出する
