Cognitive Servicesの1つに Anomaly Detector というサービスがあります。
これは時系列データの中から普段と異なる異常値を検出してくれるもので、機械学習の知識なく実装することが可能です。
今回はSynapse Analytics内のデータをAnomaly Detectorを使って異常値検出するということをやってみました。
公式手順 が準備されていますが、データ量が少ないため、もう少し大きなデータをDataLakeStorageから読み込んで利用してみます。
色々ハマったので書き残します。
Synapse Analytics作成
- 公式手順 に従いSynapse Analytics Workspaceを作成します。
- 作成できたらSynapse Studioを開きます。
Apache Sparkプール作成
-
Synapse Studioで、[管理]→[Apache Sparkプール]→[新規]の順に選択します。
-
[基本]タブを入力します。値は任意です。
-
[追加設定]では自動一時停止設定を入れています。設定は任意です。
-
[タグ]も任意です。
-
ここまで設定したらPoolを作成します。
Synapse AnalyticsとCognitive Servicesの連携
データ準備
-
今回は同様の異常検知サービスである Amazon Lookout for Metricsのサンプルデータを拝借します。
https://github.com/harunobukameda/Amazon-Lookout-for-Metrics/blob/main/ecommerce.zip -
input.csvをDataLake Storageに配置します。
タイムスタンプ確認
- ここでかなりハマりました。Anomaly Detectorではタイムスタンプ列のフォーマットがISO8601準拠である必要があります。2022-01-01T00:00:00Z の形式です。
- 事前に確認し一致していなければ変換する必要があります。
データ読み込み
-
Synapse Studioで、[開発]→[ノートブック]→[新しいノートブック]の順に選択します。
-
Jupyter Notebookで以下のように入力し、DataFrameを作成します。
from pyspark.sql import SparkSession
from pyspark.sql.types import *
account_name = "[DataLake Storage Account Name]"
container_name = "[DataLake Container Name]"
relative_path = "."
adls_path = 'abfss://%s@%s.dfs.core.windows.net/%s' % (container_name, account_name, relative_path)
df1 = spark.read.option('header', 'true') \
.option('delimiter', ',') \
.csv(adls_path + '/input.csv')
- 正しく読み込まれているか確認します。
display(df1.limit(10))
- データベース、テーブルを作成します。
%%pyspark
spark.sql("CREATE DATABASE IF NOT EXISTS [Database Name]")
df1.write.mode("overwrite").saveAsTable("[Database Name].[Table Name]")
Anomaly Detector
-
Synapse Studioより、[データ]→[レイクデータベース]→[対象DB]→[対象TABLE]を順に選択し、[Machine Learning]→[モデルを使用した予測]を選択します。
-
[Anomaly Detector]を選択します。
-
Azure Cognitive Servicesのリンクサービスには、事前に設定したリンクサービスを選択します。
-
細分性は時系列データのサンプリングレートを選択します。今回のデータは1時間毎の数値なのでhourlyを選択します。
-
タイムスタンプ列にはタイムスタンプが入った列を指定します。今回はtimestamp列を選択します。
-
時系列値列には評価するValueを選択します。今回はWebの閲覧数を表すviews列を選択します。
-
列のグループ化には時系列ごとにグルーピングする列を指定します。今回はmarketplaceを指定しています。
-
これで作成するとNotebookが開きます。
異常検知
- 開いたNotebookを以下のように編集し実行します。今回のデータでは、marketplaceをグループ化していますが、他にもplatform列を絞らないと時系列データが重複してしまうため、WHERE句を追加しています。
from mmlspark.cognitive import *
from notebookutils import mssparkutils
from pyspark.sql.functions import col
from pyspark.sql.types import DoubleType
# Load the data into a Spark DataFrame
df = spark.sql("SELECT * FROM [Database Name].[Table Name] WHERE platform = 'pc_web'").withColumn("views", col("views").cast(DoubleType()))
anomalyDetector = (SimpleDetectAnomalies()
.setLinkedService("CognitiveService1")
.setOutputCol("output")
.setErrorCol("error")
.setGranularity("hourly")
.setTimestampCol("timestamp")
.setValueCol("views")
.setGroupbyCol("marketplace"))
results = anomalyDetector.transform(df)
# Show the results
display(results.select("timestamp", "platform", "marketplace", "views", "output.*", "error"))
- 以下のように結果出力されればOKです。isAnomalyがTRUEの行が異常値と検出されています。
以上