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

Cognitive Servicesの1つに [Anomaly Detector](https://azure.microsoft.com/ja-jp/services/cognitive-services/anomaly-detector/) というサービスがあります。
これは時系列データの中から普段と異なる異常値を検出してくれるもので、機械学習の知識なく実装することが可能です。
今回はSynapse Analytics内のデータをAnomaly Detectorを使って異常値検出するということをやってみました。
[公式手順](https://docs.microsoft.com/ja-jp/azure/synapse-analytics/machine-learning/tutorial-cognitive-services-anomaly) が準備されていますが、データ量が少ないため、もう少し大きなデータをDataLakeStorageから読み込んで利用してみます。
色々ハマったので書き残します。

# Synapse Analytics作成
* [公式手順](https://docs.microsoft.com/ja-jp/azure/synapse-analytics/get-started-create-workspace) に従いSynapse Analytics Workspaceを作成します。
* 作成できたらSynapse Studioを開きます。

# Apache Sparkプール作成
* Synapse Studioで、[管理]→[Apache Sparkプール]→[新規]の順に選択します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/180335/b73cc1d6-04ab-1b23-befa-741bd2090e34.png)

* [基本]タブを入力します。値は任意です。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/180335/e49416cc-27a0-b626-4d7c-ea2979e6a51f.png)

* [追加設定]では自動一時停止設定を入れています。設定は任意です。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/180335/4d1ab33a-9973-aaab-0c83-20643725e819.png)

* [タグ]も任意です。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/180335/1de27658-7129-67f1-fd12-edb854d01cb7.png)

* ここまで設定したらPoolを作成します。

# Synapse AnalyticsとCognitive Servicesの連携
* [Azure Synapse Analytics で Cognitive Services を使用するための前提条件を構成する](https://docs.microsoft.com/ja-jp/azure/synapse-analytics/machine-learning/tutorial-configure-cognitive-services-synapse#create-an-azure-cognitive-service-linked-service-in-azure-synapse) を参考に設定します。

# データ準備
* 今回は同様の異常検知サービスである 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で、[開発]→[ノートブック]→[新しいノートブック]の順に選択します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/180335/b4c6583e-ca0e-731b-5e43-82385713c6c4.png)

* 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]→[モデルを使用した予測]を選択します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/180335/8fa143d2-c061-82b6-a468-5a8b0bb10ec3.png)

* [Anomaly Detector]を選択します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/180335/1d53514b-7e8a-a51f-5869-03b6a85c1e3c.png)

* Azure Cognitive Servicesのリンクサービスには、事前に設定したリンクサービスを選択します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/180335/15dbbd9c-9745-9bbc-5a43-77f11852d4ee.png)

* 細分性は時系列データのサンプリングレートを選択します。今回のデータは1時間毎の数値なのでhourlyを選択します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/180335/1553ffc3-a100-7b03-6359-dd68970a853b.png)

* タイムスタンプ列にはタイムスタンプが入った列を指定します。今回はtimestamp列を選択します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/180335/06152f48-f3f5-ca98-009f-6be1c55d1323.png)

* 時系列値列には評価するValueを選択します。今回はWebの閲覧数を表すviews列を選択します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/180335/6b760cbd-d690-b0e2-48cd-0148ff6d6192.png)

* 列のグループ化には時系列ごとにグルーピングする列を指定します。今回はmarketplaceを指定しています。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/180335/6b60112f-b5a8-2954-6f5d-858602a883ca.png)

* これで作成すると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の行が異常値と検出されています。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/180335/a3f0c962-1030-263e-8190-6ac65252e347.png)

以上