Oppskrift: Foundry Tools - Multivariat avviksdeteksjon

Denne oppskriften viser hvordan man bruker SynapseML og Foundry Tools på Apache Spark for multivariat avviksdeteksjon. Multivariat avviksdeteksjon innebærer deteksjon av anomalier blant mange variabler eller tidsserier, samtidig som det tas hensyn til alle innbyrdes korrelasjoner og avhengigheter mellom de forskjellige variablene. Dette scenariet bruker SynapseML og Foundry Tools for å trene en modell for deteksjon av multivariate anomalier. Deretter bruker du modellen til å utlede multivariate anomalier i et datasett som inneholder syntetiske målinger fra tre IoT-sensorer.

For mer informasjon om Azure AI Anomaly Detector, besøk informasjonsressursen Anomaly Detector.

Forutsetning

• Et Azure-abonnement – Lag et gratis
• Legg notatblokken til et lakehouse. På venstre side velger du Legg til for å legge til et eksisterende innsjøhus eller opprette et innsjøhus.

Oppsett

Fra og med en eksisterende Anomaly Detector ressurs kan du utforske måter å håndtere data i ulike former på.

Opprette en avviksdetektor ressurs

• I Azure-portalen, velg Create i ressursgruppen din, og skriv deretter Anomaly Detector. Velg den avviksdetektor ressursen.
• Gi ressursen et navn, og bruk ideelt sett samme område som resten av ressursgruppen. Bruk standardalternativene for resten, og velg deretter Se gjennom + Opprett og opprett.
• Når du har opprettet Avviksdetektor-ressursen, åpner du den og velger Keys and Endpoints panelet i venstre navigasjonsrute. Kopier nøkkelen for avviksdetektor ressursen til miljøvariabelenANOMALY_API_KEY, eller lagre den i variabelenanomalyKey.

Opprette en lagringskontoressurs

For å lagre mellomliggende data må du opprette en Azure Blob Storage-konto. Opprett en beholder for lagring av mellomliggende data i denne lagringskontoen. Noter beholdernavnet, og kopier connection string til den containeren. Du trenger den for å fylle ut variabelen containerName og miljøvariabelen BLOB_CONNECTION_STRING senere.

Skriv inn tjenestenøklene

Først, sett opp miljøvariablene for tjenestenøklene dine. Neste celle setter miljøvariablene ANOMALY_API_KEY og BLOB_CONNECTION_STRING, basert på verdiene lagret i Azure Key Vault. Hvis du kjører denne opplæringen i ditt eget miljø, må du sørge for å angi disse miljøvariablene før du fortsetter:

import os
from pyspark.sql import SparkSession
from synapse.ml.core.platform import find_secret

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

Les ANOMALY_API_KEY og BLOB_CONNECTION_STRING miljøvariablene, og angi variablene containerName og location :

# An Anomaly Detector subscription key
anomalyKey = find_secret("anomaly-api-key") # use your own anomaly api key
# Your storage account name
storageName = "anomalydetectiontest" # use your own storage account name
# A connection string to your blob storage account
storageKey = find_secret("madtest-storage-key") # use your own storage key
# A place to save intermediate MVAD results
intermediateSaveDir = (
    "wasbs://madtest@anomalydetectiontest.blob.core.windows.net/intermediateData"
)
# The location of the anomaly detector resource that you created
location = "westus2"

Koble til lagringskontoen, slik at avviksdetektoren kan lagre mellomresultater i den lagringskontoen:

spark.sparkContext._jsc.hadoopConfiguration().set(
    f"fs.azure.account.key.{storageName}.blob.core.windows.net", storageKey
)

Importer alle nødvendige moduler:

import numpy as np
import pandas as pd

import pyspark
from pyspark.sql.functions import col
from pyspark.sql.functions import lit
from pyspark.sql.types import DoubleType
import matplotlib.pyplot as plt

import synapse.ml
from synapse.ml.services import *

Les eksempeldataene inn i en Spark DataFrame:

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

df = (
    df.withColumn("sensor_1", col("sensor_1").cast(DoubleType()))
    .withColumn("sensor_2", col("sensor_2").cast(DoubleType()))
    .withColumn("sensor_3", col("sensor_3").cast(DoubleType()))
)

# Let's inspect the dataframe:
df.show(5)

Du kan nå lage et estimator objekt som du bruker til å trene modellen din. Spesifiser start- og slutttidspunktene for treningsdataene. Spesifiser også hvilke inndatakolonner som skal brukes, og navnet på kolonnen som inneholder tidsstemplene. Til slutt, spesifiser antall datapunkter som skal brukes i anomalideteksjonsvinduet, og sett connection string til Azure Blob Storage-kontoen:

trainingStartTime = "2020-06-01T12:00:00Z"
trainingEndTime = "2020-07-02T17:55:00Z"
timestampColumn = "timestamp"
inputColumns = ["sensor_1", "sensor_2", "sensor_3"]

estimator = (
    FitMultivariateAnomaly()
    .setSubscriptionKey(anomalyKey)
    .setLocation(location)
    .setStartTime(trainingStartTime)
    .setEndTime(trainingEndTime)
    .setIntermediateSaveDir(intermediateSaveDir)
    .setTimestampCol(timestampColumn)
    .setInputCols(inputColumns)
    .setSlidingWindow(200)
)

Tilpass dataene estimator til dette:

model = estimator.fit(df)

Når treningen er ferdig, bruk modellen for slutning. Koden i neste celle spesifiserer start- og slutttidspunktene for dataene der du ønsker å oppdage avvikene:

inferenceStartTime = "2020-07-02T18:00:00Z"
inferenceEndTime = "2020-07-06T05:15:00Z"

result = (
    model.setStartTime(inferenceStartTime)
    .setEndTime(inferenceEndTime)
    .setOutputCol("results")
    .setErrorCol("errors")
    .setInputCols(inputColumns)
    .setTimestampCol(timestampColumn)
    .transform(df)
)

result.show(5)

I forrige celle .show(5) vises de fem første datarammeradene. Resultatene er alle null fordi de lander utenfor slutningsvinduet.

Hvis du bare vil vise resultatene for de utledede dataene, velger du de nødvendige kolonnene. Du kan deretter ordne radene i datarammen etter stigende rekkefølge, og filtrere resultatet slik at kun radene i slutningsvinduet vises. Her inferenceEndTime matcher den siste raden i datarammen, så du kan ignorere den.

Til slutt, for å plotte resultatene bedre, konverter Spark-datarammen til en Pandas-dataramme:

rdf = (
    result.select(
        "timestamp",
        *inputColumns,
        "results.contributors",
        "results.isAnomaly",
        "results.severity"
    )
    .orderBy("timestamp", ascending=True)
    .filter(col("timestamp") >= lit(inferenceStartTime))
    .toPandas()
)

rdf

contributors Formater kolonnen som lagrer bidragspoengsummen fra hver sensor til de oppdagede avvikene. Den neste cellen håndterer dette, og deler bidragspoengsummen for hver sensor i sin egen kolonne:

def parse(x):
    if type(x) is list:
        return dict([item[::-1] for item in x])
    else:
        return {"series_0": 0, "series_1": 0, "series_2": 0}

rdf["contributors"] = rdf["contributors"].apply(parse)
rdf = pd.concat(
    [rdf.drop(["contributors"], axis=1), pd.json_normalize(rdf["contributors"])], axis=1
)
rdf

Du har nå bidragsscorene til sensorene 1, 2 og 3 i series_0henholdsvis , series_1, og series_2 kolonnene.

Hvis du vil tegne inn resultatene, kjører du neste celle. Parameteren minSeverity angir minimumsalvorlighetsgraden av avvikene som skal plottes:

minSeverity = 0.1

####### Main Figure #######
plt.figure(figsize=(23, 8))
plt.plot(
    rdf["timestamp"],
    rdf["sensor_1"],
    color="tab:orange",
    linestyle="solid",
    linewidth=2,
    label="sensor_1",
)
plt.plot(
    rdf["timestamp"],
    rdf["sensor_2"],
    color="tab:green",
    linestyle="solid",
    linewidth=2,
    label="sensor_2",
)
plt.plot(
    rdf["timestamp"],
    rdf["sensor_3"],
    color="tab:blue",
    linestyle="solid",
    linewidth=2,
    label="sensor_3",
)
plt.grid(axis="y")
plt.tick_params(axis="x", which="both", bottom=False, labelbottom=False)
plt.legend()

anoms = list(rdf["severity"] >= minSeverity)
_, _, ymin, ymax = plt.axis()
plt.vlines(np.where(anoms), ymin=ymin, ymax=ymax, color="r", alpha=0.8)

plt.legend()
plt.title(
    "A plot of the values from the three sensors with the detected anomalies highlighted in red."
)
plt.show()

####### Severity Figure #######
plt.figure(figsize=(23, 1))
plt.tick_params(axis="x", which="both", bottom=False, labelbottom=False)
plt.plot(
    rdf["timestamp"],
    rdf["severity"],
    color="black",
    linestyle="solid",
    linewidth=2,
    label="Severity score",
)
plt.plot(
    rdf["timestamp"],
    [minSeverity] * len(rdf["severity"]),
    color="red",
    linestyle="dotted",
    linewidth=1,
    label="minSeverity",
)
plt.grid(axis="y")
plt.legend()
plt.ylim([0, 1])
plt.title("Severity of the detected anomalies")
plt.show()

####### Contributors Figure #######
plt.figure(figsize=(23, 1))
plt.tick_params(axis="x", which="both", bottom=False, labelbottom=False)
plt.bar(
    rdf["timestamp"], rdf["series_0"], width=2, color="tab:orange", label="sensor_1"
)
plt.bar(
    rdf["timestamp"],
    rdf["series_1"],
    width=2,
    color="tab:green",
    label="sensor_2",
    bottom=rdf["series_0"],
)
plt.bar(
    rdf["timestamp"],
    rdf["series_2"],
    width=2,
    color="tab:blue",
    label="sensor_3",
    bottom=rdf["series_0"] + rdf["series_1"],
)
plt.grid(axis="y")
plt.legend()
plt.ylim([0, 1])
plt.title("The contribution of each sensor to the detected anomaly")
plt.show()

Plottene viser rådataene fra sensorene (inne i slutningsvinduet) i oransje, grønt og blått. De røde loddrette linjene i den første figuren viser de oppdagede avvikene som har en alvorlighetsgrad større enn eller lik minSeverity.

Det andre plottet viser alvorsgradspoengsummen for alle oppdagede avvik, med minSeverity terskelen vist i den prikkede røde linjen.

Til slutt viser det siste plottet bidraget fra dataene fra hver sensor til de oppdagede avvikene. Det hjelper med å diagnostisere og forstå den mest sannsynlige årsaken til hver avvik.

Relatert innhold

• Multivariate Anomaly Detection med Isolation Forest – krever ikke en Azure AI Anomaly Detector ressurs.
• Slik bruker du LightGBM med SynapseML
• Hvordan bruke Foundry Tools med SynapseML
• Slik bruker du SynapseML til å justere hyperparametere