Többváltozós anomáliadetektálás elkülönítési erdővel
Ez a cikk bemutatja, hogyan használható a SynapseML az Apache Sparkban a többváltozós anomáliadetektáláshoz. A többváltozós anomáliadetektálás lehetővé teszi a rendellenességek észlelését számos változó vagy időer között, figyelembe véve a különböző változók közötti összes korrelációt és függőséget. Ebben a forgatókönyvben a SynapseML használatával tanítunk be egy izolálóerdő-modellt a többváltozós anomáliadetektáláshoz, majd a betanított modellel többváltozós anomáliákat következtetünk egy olyan adathalmazon belül, amely három IoT-érzékelő szintetikus méréseit tartalmazza.
Az Elkülönítési erdő modellel kapcsolatos további információkért tekintse meg Liu et al. eredeti tanulmányát.
Előfeltételek
- Csatolja a jegyzetfüzetet egy tóházhoz. A bal oldalon válassza a Hozzáadás lehetőséget egy meglévő tóház hozzáadásához vagy egy tóház létrehozásához.
Tárimportálás
from IPython import get_ipython
from IPython.terminal.interactiveshell import TerminalInteractiveShell
import uuid
import mlflow
from pyspark.sql import functions as F
from pyspark.ml.feature import VectorAssembler
from pyspark.sql.types import *
from pyspark.ml import Pipeline
from synapse.ml.isolationforest import *
from synapse.ml.explainers import *
%matplotlib inline
from pyspark.sql import SparkSession
# Bootstrap Spark Session
spark = SparkSession.builder.getOrCreate()
from synapse.ml.core.platform import *
if running_on_synapse():
shell = TerminalInteractiveShell.instance()
shell.define_macro("foo", """a,b=10,20""")
Bemeneti adatok
# 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 training
)
inferenceEndTime = (
"2022-03-20T23:55:00Z" # datetime: datetime for when to end the training
)
# Isolation Forest parameters
contamination = 0.021
num_estimators = 100
max_samples = 256
max_features = 1.0
Adatok beolvasása
df = (
spark.read.format("csv")
.option("header", "true")
.load(
"wasbs://publicwasb@mmlspark.blob.core.windows.net/generated_sample_mvad_data.csv"
)
)
oszlopokat öntött a megfelelő adattípusokhoz
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")
)
display(df)
Betanítási adatok előkészítése
# filter to data with timestamps within the training window
df_train = df.filter(
(F.col(timestampColumn) >= trainingStartTime)
& (F.col(timestampColumn) <= trainingEndTime)
)
display(df_train)
Adatok előkészítésének tesztelése
# filter to data with timestamps within the inference window
df_test = df.filter(
(F.col(timestampColumn) >= inferenceStartTime)
& (F.col(timestampColumn) <= inferenceEndTime)
)
display(df_test)
Elkülönítési erdő modell betanítása
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)
)
Ezután létrehozunk egy ML-folyamatot az Elkülönítési erdő modell betanítása érdekében. Azt is bemutatjuk, hogyan hozhat létre MLflow-kísérletet, és regisztrálhatja a betanított modellt.
Az MLflow-modell regisztrációja szigorúan csak akkor szükséges, ha később hozzáfér a betanított modellhez. A modell betanításához és a következtetés ugyanabban a jegyzetfüzetben való végrehajtásához elegendő a modellobjektum-modell.
va = VectorAssembler(inputCols=inputCols, outputCol="features")
pipeline = Pipeline(stages=[va, isolationForest])
model = pipeline.fit(df_train)
Következtetés végrehajtása
A betanított elkülönítési erdőmodell betöltése
Következtetés végrehajtása
df_test_pred = model.transform(df_test)
display(df_test_pred)
Előre elkészített anomáliadetektor
- A legújabb pont anomáliája: létrehoz egy modellt az előző pontok használatával, és meghatározza, hogy a legújabb pont rendellenes-e (Scala, Python)
- Anomáliák keresése: egy teljes sorozatot használó modellt hoz létre, és megkeresi a sorozat rendellenességeit (Scala, Python)
Kapcsolódó tartalom
Visszajelzés
https://aka.ms/ContentUserFeedback.
Hamarosan elérhető: 2024-ben fokozatosan kivezetjük a GitHub-problémákat a tartalom visszajelzési mechanizmusaként, és lecseréljük egy új visszajelzési rendszerre. További információ:Visszajelzés küldése és megtekintése a következőhöz: