Detección de anomalías multivariantes con bosque de aislamiento

Este artículo muestra cómo usar SynapseML en Apache Spark para la detección de anomalías multivariantes. La detección de anomalías multivariantes detecta anomalías entre muchas variables o series temporales, teniendo en cuenta todas las inter-correlaciones y dependencias entre las diferentes variables. En este escenario, se utiliza SynapseML para entrenar un modelo de bosque de aislamiento para la detección de anomalías multivariantes, y luego se utiliza el modelo entrenado para inferir anomalías multivariantes dentro de un conjunto de datos que contiene mediciones sintéticas de tres sensores IoT.

Para saber más sobre el modelo del bosque de aislamiento, consulte el artículo original de Liu et al.

Prerrequisitos

1. Obtenga una suscripción a Microsoft Fabric. O bien, regístrese para obtener una evaluación gratuita de Microsoft Fabric.
2. Adjunte su bloc de notas a un lago de datos. En el lado izquierdo, seleccione Añadir para añadir un lago de datos existente o crear uno.
3. SynapseML está preinstalado en los runtimes de Fabric PySpark (se recomienda el runtime 1.3 o posterior). Para usar una versión específica, consulte Instalar una versión diferente de SynapseML en Fabric.

Importaciones de biblioteca

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()

Datos de entrada

# 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

Leer datos

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

Convierte las columnas en los tipos de datos correspondientes.

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)

Preparación de datos de entrenamiento

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

Preparación de datos de prueba

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

Entrenamiento del modelo de bosque de aislamiento

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)
)

A continuación, crea una canalización de aprendizaje automático para entrenar el modelo de Isolation Forest.

Para entrenar el modelo y realizar inferencias en el mismo cuaderno, el objeto modelo es suficiente. Para persistir y reutilizar el modelo entre sesiones, regístralo con MLflow en Microsoft Fabric.

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

Realizar inferencias

Aplica el modelo entrenado a los datos de prueba:

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

Detector de anomalías predefinido

Detector de anomalías de Azure AI

• Estado de anomalía del último punto: genera un modelo usando los puntos anteriores y determina si el último punto es anómalo. Consulta el repositorio de SynapseML en GitHub para la referencia actual de la API.
• Encontrar anomalías: genera un modelo usando una serie completa y encuentra anomalías en la serie. Consulta el repositorio de SynapseML en GitHub para la referencia actual de la API.

Contenido relacionado

• Cómo crear un motor de búsqueda con SynapseML
• Cómo usar SynapseML y las herramientas de Foundry para la detección de anomalías multivariadas: análisis de series temporales
• Uso de SynapseML para ajustar los hiperparámetros