アイソレーションフォレストを用いた多変量異常検出

この記事では、Apache Spark上でSynapseMLを使って多変量異常検出を行う方法を紹介しています。 多変量異常検出は、多くの変数や時系列間の異常を検出し、異なる変数間の相関関係や依存関係を考慮します。 このシナリオでは、SynapseMLを使って多変量異常検出用のアイソレーションフォレストモデルを訓練し、その訓練済みモデルを使って3つのIoTセンサーからの合成測定を含むデータセット内で多変量異常を推定します。

隔離森林モデルについて詳しく知りたい方は、Liuらによる元の論文をご覧ください。

[前提条件]

  1. Microsoft Fabric サブスクリプションを取得します。 または、無料の Microsoft Fabric 試用版にサインアップします。
  2. ノートブックをレイクハウスにアタッチします。 左側で [追加] を選択して、既存のレイクハウスを追加するか、レイクハウスを作成します。
  3. SynapseMLはFabric PySparkランタイム(Runtime 1.3以降推奨)にプリインストールされています。 特定のバージョンを使用するには、「FabricにSynapseMLの異なるバージョンをインストールする」をご覧ください。

ライブラリのインポート

from pyspark.sql import functions as F
from pyspark.ml.feature import VectorAssembler
from pyspark.sql.types import DoubleType
from pyspark.ml import Pipeline

from synapse.ml.isolationforest import IsolationForest

from pyspark.sql import SparkSession

# Bootstrap Spark Session
spark = SparkSession.builder.getOrCreate()

入力データ

# Table inputs
timestampColumn = "timestamp"  # str: the name of the timestamp column in the table
inputCols = [
    "sensor_1",
    "sensor_2",
    "sensor_3",
]  # list(str): the names of the input variables

# Training Start time, and number of days to use for training:
trainingStartTime = (
    "2022-02-24T06:00:00Z"  # datetime: datetime for when to start the training
)
trainingEndTime = (
    "2022-03-08T23:55:00Z"  # datetime: datetime for when to end the training
)
inferenceStartTime = (
    "2022-03-09T09:30:00Z"  # datetime: datetime for when to start the inference
)
inferenceEndTime = (
    "2022-03-20T23:55:00Z"  # datetime: datetime for when to end the inference
)

# Isolation Forest parameters
contamination = 0.021
num_estimators = 100
max_samples = 256
max_features = 1.0

データの読み取り

df = (
    spark.read.format("csv")
    .option("header", "true")
    .load(
        "wasbs://publicwasb@mmlspark.blob.core.windows.net/generated_sample_mvad_data.csv"
    )
)

列を適切なデータ型にキャストします。

df = (
    df.orderBy(timestampColumn)
    .withColumn("timestamp", F.date_format(timestampColumn, "yyyy-MM-dd'T'HH:mm:ss'Z'"))
    .withColumn("sensor_1", F.col("sensor_1").cast(DoubleType()))
    .withColumn("sensor_2", F.col("sensor_2").cast(DoubleType()))
    .withColumn("sensor_3", F.col("sensor_3").cast(DoubleType()))
    .drop("_c5")  # drop the extra unlabeled column present in the source CSV
)

display(df)

トレーニング データの準備

# filter to data with timestamps within the training window
df_train = df.filter(
    (F.col(timestampColumn) >= trainingStartTime)
    & (F.col(timestampColumn) <= trainingEndTime)
)
display(df_train)

テストデータの準備

# filter to data with timestamps within the inference window
df_test = df.filter(
    (F.col(timestampColumn) >= inferenceStartTime)
    & (F.col(timestampColumn) <= inferenceEndTime)
)
display(df_test)

アイソレーションフォレストモデルの学習

isolationForest = (
    IsolationForest()
    .setNumEstimators(num_estimators)
    .setBootstrap(False)
    .setMaxSamples(max_samples)
    .setMaxFeatures(max_features)
    .setFeaturesCol("features")
    .setPredictionCol("predictedLabel")
    .setScoreCol("outlierScore")
    .setContamination(contamination)
    .setContaminationError(0.01 * contamination)
    .setRandomSeed(1)
)

次に、Isolation Forestモデルを訓練するためのMLパイプラインを作成します。

モデルの訓練と推論を行う場合、同じノートブック上でモデルオブジェクトで十分です。 モデルをセッション間で永続化・再利用するには、Microsoft FabricのMLflowで登録してください。

va = VectorAssembler(inputCols=inputCols, outputCol="features")
pipeline = Pipeline(stages=[va, isolationForest])
model = pipeline.fit(df_train)

推論を実行する

訓練済みモデルをテストデータに適用します:

df_test_pred = model.transform(df_test)
display(df_test_pred)

事前に作成されたアノマリーデテクター

Important

Microsoftは2026年10月1日にAzure AI Anomaly Detectorサービスを終了します。 2023年9月20日以降、新しいリソースを作成できなくなりました。 サポートされた代替案については、Microsoft Fabric Real-Timeインテリジェンスの異常検出を参照してください。

Azure AI Anomaly Detector

  • 最新点の異常状態:前の点を用いてモデルを生成し、最新の点が異常かどうかを判定します。 現在のAPI参照はSynapseMLのGitHubリポジトリをご覧ください。
  • 異常を見つける:一連のモデルを生成し、その系列の異常を見つけます。 現在のAPI参照はSynapseMLのGitHubリポジトリをご覧ください。

関連するコンテンツ

  • Last updated on