Detecção de anomalias multivariadas

Este tutorial mostra como treinar um modelo de detecção de anomalias multivariadas em um bloco de anotações Fabric usando dados de exemplo armazenados em uma casa de eventos. Em seguida, use o modelo treinado em um conjunto de consultas KQL para pontuar novos dados e visualizar anomalias.

Para obter informações em segundo plano, consulte Detecção de anomalias multivariadas em Microsoft Fabric – visão geral.

Pré-requisitos

Parte 1: Ativar a disponibilidade do OneLake

Ative a disponibilidade do OneLake antes de carregar dados na casa de eventos. Essa configuração disponibiliza os dados ingeridos no OneLake para que você possa acessar a mesma tabela em um notebook mais adiante no tutorial.

  1. Em seu workspace, abra a casa de eventos que você criou nos pré-requisitos e selecione o banco de dados no qual deseja armazenar seus dados.

  2. No painel de detalhes do Banco de Dados , defina a disponibilidade do OneLake como Ativada.

    Captura de tela da habilitação da disponibilidade do OneLake em sua casa de eventos.

Parte 2: Ative o plug-in KQL do Python

Nesta etapa, você ativa o plug-in Python em sua casa de eventos. Esta etapa é necessária para executar o código Python no conjunto de consultas KQL na Parte 9: prever anomalias em um conjunto de consultas KQL. Selecione a imagem do Python que inclui o pacote time-series-anomaly-detector.

  1. No Eventhouse, selecione Eventhouse>Plugins na faixa de opções.

  2. No painel Plug-ins, defina Python extensão de idioma como Ativado.

  3. Selecione Python DL 3.11.7.

  4. Selecione Concluído.

    Captura de tela mostrando como habilitar o pacote de DL do Python 3.11.7 na casa de eventos.

Parte 3: Criar um ambiente spark

Nesta etapa, você criará um ambiente spark para executar o notebook que treina o modelo de detecção de anomalias multivariadas. Para obter mais informações, consulte Criar e gerenciar ambientes.

  1. No seu espaço de trabalho, selecione + Novo item e, em seguida, selecione Ambiente.

    Captura de tela do bloco Ambiente na janela Novo item.

  2. Insira MVAD_ENV o nome do ambiente e selecione Criar.

  3. Em Bibliotecas, selecione Bibliotecas públicas.

  4. Selecione Adicionar do PyPI.

  5. Na caixa de pesquisa, insira time-series-anomaly-detector. Na caixa Versão, insira0.3.9.

  6. Selecione Salvar.

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

  7. Selecione a guia Início no ambiente.

  8. Na faixa de opções, selecione o ícone Publicar.

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

    Captura de tela da publicação do ambiente.

Parte 4: Carregar dados no Eventhouse

  1. Na casa de eventos, passe o mouse sobre o banco de dados KQL onde você deseja armazenar seus dados e selecione Mais menu [...]>Obter dados>Arquivo local.

    Captura de tela de obtenção de dados de um arquivo local.

  2. Selecione + Nova tabela e insira demo_stocks_change como o nome da tabela.

  3. Na caixa de diálogo de upload, selecione Procurar arquivos e carregue o arquivo de dados de exemplo baixado em Pré-requisitos.

  4. Selecione Avançar.

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

  6. Selecione Concluir.

  7. Após a conclusão do carregamento, selecione Fechar.

Parte 5: Copiar o caminho do OneLake

Selecione a demo_stocks_change tabela. No painel Detalhes da tabela, selecione Pasta do OneLake para copiar o caminho do OneLake para a área de transferência. Salve o caminho em um editor de texto para uso posterior.

Captura de tela da cópia do caminho do OneLake.

Parte 6: Preparar o bloco de anotações

  1. Selecione o espaço de trabalho.

  2. Selecione Importar>Bloco de Anotações>deste computador.

  3. Selecione Carregar e escolha o notebook que você baixou em Pré-requisitos.

  4. Depois que o notebook for carregado, você poderá encontrar e abrir seu notebook no espaço de trabalho.

  5. Na faixa de opções superior, selecione a lista suspensa padrão do Workspace e selecione o ambiente que você criou na etapa anterior.

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

Parte 7: Executar o notebook

  1. Importe os pacotes padrão.

    import numpy as np
import pandas as pd

  2. O Spark precisa de um URI do ABFSS para se conectar com segurança ao armazenamento do OneLake, portanto, defina uma função auxiliar que converte o URI do OneLake em um URI do 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 do OneLake copiado na Parte 5: copie o caminho do OneLake e carregue a demo_stocks_change tabela em um dataframe do 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 células a seguir para preparar os dataframes de treinamento e previsão.

    Observação

    As previsões propriamente ditas são executadas no Eventhouse descrito em Parte 9: Prever anomalias em uma consulta KQL. Em um cenário de produção, você normalmente pontua novos dados de streaming. Neste tutorial, o conjunto de dados é dividido por data em intervalos de treinamento e previsão para simular dados históricos e de entrada.

    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 salvá-lo no registro de modelo do 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 célula a seguir para obter o caminho do modelo registrado que você usará mais tarde para fazer previsões no sandbox de 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 da saída da última célula. Você o usa na Parte 9.

Parte 8: Criar um conjunto de consultas KQL

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

  1. No seu espaço de trabalho, selecione + Novo item>KQL Queryset.
  2. Insira MultivariateAnomalyDetectionTutoriale selecione Criar.
  3. Na janela do catálogo do OneLake , selecione o banco de dados KQL em que você armazenou os dados.
  4. Selecione Conectar.

Parte 9: Prever anomalias em um 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 consulta de previsão a seguir. Substitua enter your model URI here pelo URI que você copiou ao final da Parte 7: Execute o notebook.

    A consulta detecta anomalias multivariadas entre os cinco estoques usando o modelo treinado e renderiza os resultados como um anomalychart. Os pontos anômalos são exibidos na primeira ação (AAPL), mas representam anomalias no comportamento conjunto de todas as cinco ações em uma determinada 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 se assemelha à seguinte imagem:

Captura de tela da saída de anomalia multivariada.

Limpar os recursos

Ao concluir o tutorial, exclua os recursos criados para evitar custos desnecessários:

  1. Navegue até a home page do espaço de trabalho.
  2. Exclua o ambiente criado neste tutorial.
  3. Exclua o notebook criado neste tutorial.
  4. Exclua a casa de eventos ou o banco de dados usado neste tutorial.
  5. Exclua o conjunto de consultas KQL criado neste tutorial.

Conteúdo relacionado

  • Last updated on