Azure Synapse Analytics + Anomaly Detectorで異常値検出をやってみる

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の連携

• Azure 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の行が異常値と検出されています。

以上