Deteção de anomalias multivariável

Este tutorial mostra-lhe como treinar um modelo multivariado de deteção de anomalias num caderno Fabric, utilizando dados de amostra armazenados numa casa de eventos. Depois, utiliza-se o modelo treinado num conjunto de consultas KQL para obter novos dados e visualizar anomalias.

Para informações de contexto, veja Deteção de anomalias multivariadas no Microsoft Fabric - visão geral.

Pré-requisitos

Parte 1: Ativar a disponibilidade do OneLake

Ativa a disponibilidade do OneLake antes de carregares dados na casa de eventos. Esta configuração torna os dados ingeridos disponíveis no OneLake para que possas aceder à mesma tabela a partir de um caderno mais tarde no tutorial.

  1. No teu espaço de trabalho, abre a casa de eventos que criaste nos pré-requisitos e depois seleciona a base de dados onde queres guardar os teus dados.

  2. No painel de detalhes da base de dados , defina a disponibilidade do OneLake para Ligado.

    Captura de ecrã de ativar a disponibilidade do OneLake na tua casa de eventos.

Parte 2: Ativar o plugin KQL Python

Neste passo, ativas o plugin Python na tua casa de eventos. Este passo é necessário para executar o código Python no conjunto de consultas KQL na Parte 9: Prever anomalias num conjunto de consultas KQL. Selecione a imagem Python que inclui o pacote time-series-anomaly-detector.

  1. Na casa de eventos, selecione Plugins da Casa de Eventos> na fita.

  2. No painel Extensões, defina extensão da linguagem Python para Ativado.

  3. Selecione Python 3.11.7 DL.

  4. Selecionar Concluído.

    Captura de ecrã a mostrar como ativar o pacote DL do Python 3.11.7 no eventhouse.

Parte 3: Criar um ambiente Spark

Neste passo, cria um ambiente Spark para executar o caderno que treina o modelo de deteção multivariada de anomalias. Para mais informações, consulte Criar e gerir ambientes.

  1. No teu espaço de trabalho, seleciona + Novo item e depois seleciona Ambiente.

    Captura de ecrã do mosaico Ambiente na janela de Novo Item.

  2. Inscreva MVAD_ENV o nome do ambiente e depois selecione Criar.

  3. Em Bibliotecas, selecione Bibliotecas públicas.

  4. Selecionar Adicionar a partir do PyPI.

  5. Na caixa de pesquisa, digite time-series-anomaly-detector. Na caixa Versão , introduza 0.3.9.

  6. Selecione Guardar.

    Captura de tela da adição do pacote PyPI ao ambiente Spark.

  7. Selecione a aba Início no ambiente.

  8. Selecione o ícone Publicar na barra de ferramentas.

  9. Selecione Publicar tudo. Esta etapa pode levar vários minutos para ser concluída.

    Captura de ecrã da publicação do ambiente.

Parte 4: Carregar os dados no eventhouse

  1. Na casa de eventos, passe o rato sobre a base de dados KQL onde pretende guardar os seus dados e depois selecione Mais menu [...]>Obter dados>Ficheiro local.

    Captura de ecrã de obtenção de dados a partir de ficheiro local.

  2. Selecionar + Nova tabela e inserir demo_stocks_change como nome da tabela.

  3. No diálogo de upload, selecione Procurar ficheiros e carregue o ficheiro de dados de exemplo que descarregou em Pré-requisitos.

  4. Selecione Seguinte.

  5. Na secção Inspecionar os dados, verifique se Primeira linha como cabeçalho da coluna está definido como Ativado.

  6. Selecione Concluir.

  7. Quando os dados forem carregados, selecione Fechar.

Parte 5: Copiar o caminho OneLake

Selecione a demo_stocks_change tabela. No painel de detalhes da tabela , selecione a pasta OneLake para copiar o caminho OneLake para a sua prancheta. Guarde o caminho num editor de texto para uso posterior.

Captura de ecrã de copiar o caminho do OneLake.

Parte 6: Preparar o caderno

  1. Selecione a área de trabalho.

  2. Selecione Importar>Bloco de Notas>a partir deste computador.

  3. Seleciona Carregar e escolhe o caderno que descarregaste em Pré-requisitos.

  4. Depois de carregar o bloco de notas, pode localizar e abrir o bloco de notas a partir da sua área de trabalho.

  5. No friso superior, selecione a lista pendente predefinição do Workspace e, em seguida, selecione o ambiente que criou na etapa anterior.

    Captura de ecrã da seleção do ambiente no notebook.

Parte 7: Executar o caderno

  1. Importar pacotes padrão.

    import numpy as np
import pandas as pd

  2. O Spark precisa de um URI ABFSS para se ligar de forma segura ao armazenamento OneLake, por isso defina uma função auxiliar que converta o URI OneLake num URI ABFSS.

    def convert_onelake_to_abfss(onelake_uri):
    if not onelake_uri.startswith('https://'):
        raise ValueError("Invalid OneLake URI. It should start with 'https://'.")
    uri_without_scheme = onelake_uri[8:]
    parts = uri_without_scheme.split('/')
    if len(parts) < 3:
        raise ValueError("Invalid OneLake URI format.")
    container_name = parts[1]
    path = '/'.join(parts[2:])
    abfss_uri = f"abfss://{container_name}@{parts[0]}/{path}"
    return abfss_uri

  3. Substitua OneLakeTableURI pelo URI OneLake que copiou na Parte 5: Copie o caminho OneLake e depois carregue a demo_stocks_change tabela num dataframe pandas.

    onelake_uri = "OneLakeTableURI"  # Replace with your OneLake table URI.
abfss_uri = convert_onelake_to_abfss(onelake_uri)
print(abfss_uri)

    df = spark.read.format('delta').load(abfss_uri)
df = df.toPandas()
df['Date'] = pd.to_datetime(df['Date'])
df = df.set_index('Date').sort_index()
print(df.shape)
df.head(3)

  4. Execute as seguintes células para preparar os dataframes de treino e previsão.

    Nota

    As previsões reais são executadas no eventhouse descrito na Parte 9: Prever anomalias num conjunto de consultas em KQL. Num cenário de produção, normalmente conseguimos novos dados de streaming. Neste tutorial, o conjunto de dados é dividido por datas em intervalos de treino e de previsão para simular dados históricos e novos dados.

    features_cols = ['AAPL', 'AMZN', 'GOOG', 'MSFT', 'SPY']
cutoff_date = pd.Timestamp('2023-01-01')

    train_df = df.loc[df.index < cutoff_date, features_cols]
print(train_df.shape)
train_df.head(3)

    train_len = len(train_df)
predict_len = len(df) - train_len
print(f'Total samples: {len(df)}. Split to {train_len} for training, {predict_len} for testing')

  5. Execute as células para treinar o modelo e guarde-o no registo do modelo Fabric MLflow.

    from anomaly_detector import MultivariateAnomalyDetector
model = MultivariateAnomalyDetector()

    sliding_window = 200
params = {"sliding_window": sliding_window}

    model.fit(train_df, params=params)

    model_name = "mvad_5_stocks_model"

    import mlflow

with mlflow.start_run():
    mlflow.log_params(params)
    mlflow.set_tag("Training Info", "MVAD on 5 Stocks Dataset")

    model_info = mlflow.pyfunc.log_model(
        python_model=model,
        artifact_path="mvad_artifacts",
        registered_model_name=model_name,
    )

  6. Execute a seguinte célula para obter o caminho do modelo registado, que utilizará mais tarde para fazer previsões no sandbox Python do KQL.

    from mlflow.tracking import MlflowClient

client = MlflowClient()
mvs = client.search_model_versions(f"name='{model_name}'")
latest = max(mvs, key=lambda v: v.creation_timestamp)
model_abfss = latest.source
print(model_abfss)

  7. Copie o URI do modelo a partir da saída da última célula. Usas isso na Parte 9.

Parte 8: Criar um conjunto de consultas KQL

Para obter informações gerais, consulte Criar um conjunto de consultas KQL.

  1. No seu espaço de trabalho, selecione + Novo item>KQL Queryset.
  2. Introduza MultivariateAnomalyDetectionTutorial, e depois selecione Criar.
  3. Na janela do catálogo OneLake , selecione a base de dados KQL onde armazenou os dados.
  4. Selecione Ligar.

Parte 9: Prever anomalias num conjunto de consultas KQL

  1. Execute a seguinte .create-or-alter function consulta para definir a predict_fabric_mvad_fl() função armazenada:

    .create-or-alter function with (folder = "Packages\\ML", docstring = "Predict MVAD model in Microsoft Fabric")
predict_fabric_mvad_fl(samples:(*), features_cols:dynamic, artifacts_uri:string, trim_result:bool=false)
{
    let s = artifacts_uri;
    let artifacts = bag_pack('MLmodel', strcat(s, '/MLmodel;impersonate'), 'conda.yaml', strcat(s, '/conda.yaml;impersonate'),
                             'requirements.txt', strcat(s, '/requirements.txt;impersonate'), 'python_env.yaml', strcat(s, '/python_env.yaml;impersonate'),
                             'python_model.pkl', strcat(s, '/python_model.pkl;impersonate'));
    let kwargs = bag_pack('features_cols', features_cols, 'trim_result', trim_result);
    let code = ```if 1:
        import os
        import shutil
        import mlflow
        work_dir = os.environ.get("UPLOAD_PATH")
        model_dir = work_dir + '/mvad_model'
        model_data_dir = model_dir + '/data'
        os.mkdir(model_dir)
        shutil.move(work_dir + '/MLmodel', model_dir)
        shutil.move(work_dir + '/conda.yaml', model_dir)
        shutil.move(work_dir + '/requirements.txt', model_dir)
        shutil.move(work_dir + '/python_env.yaml', model_dir)
        shutil.move(work_dir + '/python_model.pkl', model_dir)
        features_cols = kargs["features_cols"]
        trim_result = kargs["trim_result"]
        test_data = df[features_cols]
        model = mlflow.pyfunc.load_model(model_dir)
        predictions = model.predict(test_data)
        predict_result = pd.DataFrame(predictions)
        samples_offset = len(df) - len(predict_result)        # this model doesn't output predictions for the first sliding_window-1 samples
        if trim_result:                                       # trim the prefix samples
            result = df[samples_offset:]
            result.iloc[:,-4:] = predict_result.iloc[:, 1:]   # no need to copy 1st column which is the timestamp index
        else:
            result = df                                       # output all samples
            result.iloc[samples_offset:,-4:] = predict_result.iloc[:, 1:]
        ```;
    samples
    | evaluate python(typeof(*), code, kwargs, external_artifacts=artifacts)
}

  2. Execute a seguinte consulta de previsão. Substitui enter your model URI here pelo URI que copiaste no final da Parte 7: Executa o caderno.

    A consulta deteta anomalias multivariadas nas cinco ações usando o modelo treinado e, em seguida, apresenta os resultados como um anomalychart. Os pontos anómalos são exibidos no primeiro stock (AAPL), mas representam anomalias no comportamento conjunto de todos os cinco stocks numa dada data.

    let cutoff_date=datetime(2023-01-01);
let num_predictions=toscalar(demo_stocks_change | where Date >= cutoff_date | count);   //  number of latest points to predict
let sliding_window=200;                                                                 //  should match the window that was set for model training
let prefix_score_len = sliding_window/2+min_of(sliding_window/2, 200)-1;
let num_samples = prefix_score_len + num_predictions;
demo_stocks_change
| top num_samples by Date desc
| order by Date asc
| extend is_anomaly=bool(false), score=real(null), severity=real(null), interpretation=dynamic(null)
| invoke predict_fabric_mvad_fl(pack_array('AAPL', 'AMZN', 'GOOG', 'MSFT', 'SPY'),
            // NOTE: Update artifacts_uri to model path
            artifacts_uri='enter your model URI here',
            trim_result=true)
| summarize Date=make_list(Date), AAPL=make_list(AAPL), AMZN=make_list(AMZN), GOOG=make_list(GOOG), MSFT=make_list(MSFT), SPY=make_list(SPY), anomaly=make_list(toint(is_anomaly))
| render anomalychart with(anomalycolumns=anomaly, title='Stock price changes in % with anomalies')

O gráfico de anomalias resultante assemelha-se à seguinte imagem:

Captura de ecrã da saída de anomalia multivariada.

Limpar recursos

Quando terminares o tutorial, elimina os recursos que criaste para evitar custos desnecessários:

  1. Navegue até a página inicial do seu espaço de trabalho.
  2. Exclua o ambiente criado neste tutorial.
  3. Exclua o bloco de anotações criado neste tutorial.
  4. Apaga a casa de eventos ou a base de dados usada neste tutorial.
  5. Exclua o conjunto de consultas KQL criado neste tutorial.

Conteúdo relacionado

  • Last updated on