Tutorial: criar, avaliar e pontuar um modelo de detecção de falhas do computador

Este tutorial apresenta um exemplo de ponta a ponta de um fluxo de trabalho da Ciência de Dados do Synapse no Microsoft Fabric. O cenário usa o aprendizado de máquina para ter uma abordagem mais sistemática para o diagnóstico de falhas a fim de identificar proativamente problemas e tomar ações antes da falha real de um computador. O objetivo é prever se um computador sofreria uma falha com base na temperatura do processo, velocidade rotacional e etc.

Este tutorial cobre estas etapas:

Pré-requisitos

• Obtenha uma assinatura do Microsoft Fabric. Ou, inscreva-se para uma avaliação gratuita do Microsoft Fabric.

• Entre no Microsoft Fabric.

• Use o seletor de experiência no lado esquerdo da sua página inicial para alternar para a experiência de Ciência de Dados do Synapse.

  

• Se necessário, crie um lakehouse do Microsoft Fabric como está descrito em Criar um lakehouse no Microsoft Fabric.

Acompanhar em um notebook

Você pode escolher uma dessas opções para acompanhar em um notebook:

• Abra e execute o Notebook interno na experiência de Ciência de Dados
• Carregue seu Notebook do GitHub para a experiência de Ciência de Dados

Abrir o Notebook interno

O notebook de exemplo Falha do computador acompanha este tutorial.

Para abrir o Notebook de amostra interno do tutorial na experiência de Ciência de Dados do Synapse:

1. Vá para a página inicial de Synapse Data Science.

2. Selecione Usar um exemplo.

3. Selecione o exemplo correspondente:

   • Na guia padrão fluxos de trabalho de ponta a ponta (Python), se o exemplo for para um tutorial do Python.
   • Na guia fluxos de trabalho de ponta a ponta (R), se o exemplo for para um tutorial do R.
   • Na guia Tutoriais rápidos, se o exemplo for para um tutorial rápido.

4. Anexe um lakehouse ao notebook antes de começar a executar o código.

Importar o Notebook do GitHub

O notebook AISample – Manutenção Preditiva acompanha este tutorial.

Para abrir o notebook que acompanha este tutorial, siga as instruções em Preparar seu sistema para ciência de dados para importar os notebooks do tutorial para seu espaço de trabalho.

Se preferir copiar e colar o código a partir dessa página, você poderá criar um novo notebook.

Certifique-se de anexar um lakehouse ao notebook antes de começar a executar o código.

Etapa 1: Instalar bibliotecas personalizadas

Para o desenvolvimento de modelo de machine learning ou análise de dados ad-hoc, talvez seja necessário instalar rapidamente uma biblioteca personalizada para sua sessão do Apache Spark. Você tem duas opções para instalar bibliotecas.

• Usar os recursos de instalação embutido (%pip ou %conda) do notebook para instalar a biblioteca somente no notebook atual.
• Como alternativa, você pode criar um ambiente de Fabric, instalar bibliotecas de fontes públicas ou carregar bibliotecas personalizadas para ele e, em seguida, o administrador do espaço de trabalho pode anexar o ambiente como padrão para o espaço de trabalho. Todas as bibliotecas no ambiente ficarão disponíveis para uso em quaisquer notebooks e definições de trabalho do Spark no espaço de trabalho. Para obter mais informações sobre ambientes, confira Criar, configurar e usar um ambiente no Microsoft Fabric.

Para este tutorial, use %pip install para instalar a biblioteca imblearn no notebook.

# Use pip to install imblearn
%pip install imblearn

Etapa 2: Carregar os dados

O conjunto de dados simula o registro em log dos parâmetros de uma máquina de fabricação como uma função do tempo, o que é comum nas configurações industriais. Consiste em 10.000 pontos de dados armazenados como linhas com recursos como colunas. Os recursos incluem:

• Um UID (Identificador Exclusivo) que varia de 1 a 10.000

• ID do produto, que consiste em uma letra L (para baixa), M (para média) ou H (para alta), que indica a variante de qualidade do produto e um número de série específico da variante. As variantes de baixa, média e alta qualidades compõem 60%, 30% e 10% de todos os produtos, respectivamente

• Temperatura do ar, em graus Kelvin (K)

• Temperatura de processo, em graus Kelvin

• Velocidade de rotação, em rotações por minuto (RPM)

• Torque, em Newton-metros (Nm)

• Desgaste da ferramenta, em minutos. As variantes de qualidade H, M e L adicionam 5, 3 e 2 minutos de desgaste da ferramenta respectivamente à ferramenta usada no processo

• Um rótulo de falha do computador, para indicar se a máquina falhou no ponto de dados específico. Esse ponto de dados específico pode ter qualquer um dos cinco modos de falha independentes a seguir:

  • Falha de Desgaste da Ferramenta (TWF): a ferramenta é substituída ou falha em um tempo de desgaste da ferramenta selecionado aleatoriamente entre 200 e 240 minutos
  • Falha de Dissipação de Calor (HDF): a dissipação de calor causa uma falha no processo se a diferença entre o ar e a temperatura do processo é menor que 8,6 K e a velocidade de rotação da ferramenta é menor que 1.380 RPM
  • Falha de Energia (PWF): o produto do torque e a velocidade de rotação (em rad/s) são iguais à potência necessária para o processo. O processo falhará se essa energia cair para menos de 3.500 W ou exceder 9.000 W
  • Falha de Sobrecarga (OSF): se o produto do desgaste da ferramenta e do torque excede 11.000 Nm mínimos para a variante do produto L (12.000 M, 13.000 H), o processo falha devido à sobrecarga
  • Falhas Aleatórias (RNF): cada processo tem uma chance de falha de 0,1%, independentemente dos parâmetros do processo

Baixar o conjunto de dados e carregar no lakehouse

Conecte-se ao Contêiner do Azure Open Datasets e carregue o conjunto de dados da Manutenção Preditiva. Esse código faz o download de uma versão disponível publicamente do conjunto de dados e, em seguida, o armazena em um lakehouse do Fabric:

# Download demo data files into the lakehouse if they don't exist
import os, requests
DATA_FOLDER = "Files/predictive_maintenance/"  # Folder that contains the dataset
DATA_FILE = "predictive_maintenance.csv"  # Data file name
remote_url = "https://synapseaisolutionsa.blob.core.windows.net/public/MachineFaultDetection"
file_list = ["predictive_maintenance.csv"]
download_path = f"/lakehouse/default/{DATA_FOLDER}/raw"

if not os.path.exists("/lakehouse/default"):
    raise FileNotFoundError(
        "Default lakehouse not found, please add a lakehouse and restart the session."
    )
os.makedirs(download_path, exist_ok=True)
for fname in file_list:
    if not os.path.exists(f"{download_path}/{fname}"):
        r = requests.get(f"{remote_url}/{fname}", timeout=30)
        with open(f"{download_path}/{fname}", "wb") as f:
            f.write(r.content)
print("Downloaded demo data files into lakehouse.")

Depois que o conjunto de dados for baixado para o lakehouse, você poderá carregá-lo como um DataFrame do Spark:

df = (
    spark.read.option("header", True)
    .option("inferSchema", True)
    .csv(f"{DATA_FOLDER}raw/{DATA_FILE}")
    .cache()
)
df.show(5)

Esta tabela mostra uma amostra de visualização dos dados:

Gravar um DataFrame do Spark na tabela delta do lakehouse

Formate os dados (por exemplo, substitua os espaços por sublinhados) para facilitar as operações do Spark nas etapas subsequentes:

# Replace the space in the column name with an underscore to avoid an invalid character while saving 
df = df.toDF(*(c.replace(' ', '_') for c in df.columns))
table_name = "predictive_maintenance_data"
df.show(5)

Esta tabela mostra uma visualização dos dados com nomes de coluna reformatados:

# Save data with processed columns to the lakehouse 
df.write.mode("overwrite").format("delta").save(f"Tables/{table_name}")
print(f"Spark DataFrame saved to delta table: {table_name}")

Etapa 3: Pré-processar dados e executar análise exploratória dos dados

Converta o DataFrame do Spark em DataFrame do Pandas para usar as bibliotecas de gráficos populares compatíveis com o Pandas.

data = spark.read.format("delta").load("Tables/predictive_maintenance_data")
SEED = 1234
df = data.toPandas()
df.drop(['UDI', 'Product_ID'],axis=1,inplace=True)
# Rename the Target column to IsFail
df = df.rename(columns = {'Target': "IsFail"})
df.info()

Converta as colunas específicas do conjunto de dados para floats ou tipos inteiros como requeridos e cadeias de caracteres de mapa, como ('L', 'M', 'H'), em valores numéricos (0, 1, 2):

# Convert temperature, rotational speed, torque, and tool wear columns to float
df['Air_temperature_[K]'] = df['Air_temperature_[K]'].astype(float)
df['Process_temperature_[K]'] = df['Process_temperature_[K]'].astype(float)
df['Rotational_speed_[rpm]'] = df['Rotational_speed_[rpm]'].astype(float)
df['Torque_[Nm]'] = df['Torque_[Nm]'].astype(float)
df['Tool_wear_[min]'] = df['Tool_wear_[min]'].astype(float)

# Convert the 'Target' column to an integer 
df['IsFail'] = df['IsFail'].astype(int)
# Map 'L', 'M', 'H' to numerical values 
df['Type'] = df['Type'].map({'L': 0, 'M': 1, 'H': 2})

Explorar dados por meio de visualizações

# Import packages and set plotting style
import seaborn as sns
import matplotlib.pyplot as plt
import pandas as pd
sns.set_style('darkgrid')

# Create the correlation matrix
corr_matrix = df.corr(numeric_only=True)

# Plot a heatmap
plt.figure(figsize=(10, 8))
sns.heatmap(corr_matrix, annot=True)
plt.show()

Conforme esperado, a falha (IsFail) correlaciona com os recursos selecionados (colunas). A matriz de correlação mostra que Air_temperature, Process_temperature, Rotational_speed, Torque e Tool_wear têm a correlação mais alta com a variável IsFail.

# Plot histograms of select features
fig, axes = plt.subplots(2, 3, figsize=(18,10))
columns = ['Air_temperature_[K]', 'Process_temperature_[K]', 'Rotational_speed_[rpm]', 'Torque_[Nm]', 'Tool_wear_[min]']
data=df.copy()
for ind, item in enumerate (columns):
    column = columns[ind]
    df_column = data[column]
    df_column.hist(ax = axes[ind%2][ind//2], bins=32).set_title(item)
fig.supylabel('count')
fig.subplots_adjust(hspace=0.2)
fig.delaxes(axes[1,2])

Como mostram os gráficos plotados, as variáveis Air_temperature, Process_temperature, Rotational_speed, Torque, e Tool_wear não são esparsas. Eles parecem ter boa continuidade no espaço de recursos. Esses gráficos confirmam que o treinamento de um modelo de machine learning nesse conjunto de dados provavelmente produz resultados confiáveis e que podem ser generalizados para um novo conjunto de dados.

Inspecionar a variável de destino quanto ao desequilíbrio de classe

Conte o número de amostras para os computadores com e sem falha e inspecione o equilíbrio dos dados para cada classe (IsFail=0, IsFail=1):

# Plot the counts for no failure and each failure type
plt.figure(figsize=(12, 2))
ax = sns.countplot(x='Failure_Type', data=df)
for p in ax.patches:
    ax.annotate(f'{p.get_height()}', (p.get_x()+0.4, p.get_height()+50))

plt.show()

# Plot the counts for no failure versus the sum of all failure types
plt.figure(figsize=(4, 2))
ax = sns.countplot(x='IsFail', data=df)
for p in ax.patches:
    ax.annotate(f'{p.get_height()}', (p.get_x()+0.4, p.get_height()+50))

plt.show()

Os gráficos indicam que a classe sem falha (mostrada como IsFail=0 no segundo gráfico) constitui a maioria das amostras. Use uma técnica de sobreamostragem para criar um conjunto de dados de treinamento mais equilibrado:

# Separate features and target
features = df[['Type', 'Air_temperature_[K]', 'Process_temperature_[K]', 'Rotational_speed_[rpm]', 'Torque_[Nm]', 'Tool_wear_[min]']]
labels = df['IsFail']

# Split the dataset into the training and testing sets
from sklearn.model_selection import train_test_split
X_train, X_test, y_train, y_test = train_test_split(features, labels, test_size=0.2, random_state=42)

# Ignore warnings
import warnings
warnings.filterwarnings('ignore')
# Save test data to the lakehouse for use in future sections
table_name = "predictive_maintenance_test_data"
df_test_X = spark.createDataFrame(X_test)
df_test_X.write.mode("overwrite").format("delta").save(f"Tables/{table_name}")
print(f"Spark DataFrame saved to delta table: {table_name}")

Sobreamostragem para equilibrar as classes no conjunto de dados de treinamento

A análise anterior mostrou que o conjunto de dados está altamente desequilibrado. Esse desequilíbrio se torna um problema, porque há poucos exemplos da classe minoritária para que um modelo aprenda efetivamente o limite de decisão.

SMOTE pode resolver o problema. O SMOTE é uma técnica de sobreamostragem amplamente usada que gera exemplos sintéticos. Ele gera exemplos para a classe minoritária com base nas distâncias euclidianas entre os pontos de dados. Esse método difere da sobreamostragem aleatória, porque cria novos exemplos que não apenas duplicam a classe minoritária. O método torna-se uma técnica mais eficaz para lidar com conjuntos de dados desequilibrados.

# Disable MLflow autologging because you don't want to track SMOTE fitting
import mlflow

mlflow.autolog(disable=True)

from imblearn.combine import SMOTETomek
smt = SMOTETomek(random_state=SEED)
X_train_res, y_train_res = smt.fit_resample(X_train, y_train)

# Plot the counts for both classes
plt.figure(figsize=(4, 2))
ax = sns.countplot(x='IsFail', data=pd.DataFrame({'IsFail': y_train_res.values}))
for p in ax.patches:
    ax.annotate(f'{p.get_height()}', (p.get_x()+0.4, p.get_height()+50))

plt.show()

Você equilibrou com êxito o conjunto de dados. Agora você pode passar para o treinamento de modelo.

Etapa 4: Treinar e avaliar os modelos

O MLflow registra modelos, treina e compara vários modelos e escolhe o melhor para fins de previsão. Use os três modelos a seguir para o treinamento do modelo:

• Classificador de floresta aleatória
• Classificador de regressão logística
• Classificador XGBoost

Treinar classificador de floresta aleatória

import numpy as np 
from sklearn.ensemble import RandomForestClassifier
from mlflow.models.signature import infer_signature
from sklearn.metrics import f1_score, accuracy_score, recall_score

mlflow.set_experiment("Machine_Failure_Classification")
mlflow.autolog(exclusive=False) # This is needed to override the preconfigured autologging behavior

with mlflow.start_run() as run:
    rfc_id = run.info.run_id
    print(f"run_id {rfc_id}, status: {run.info.status}")
    rfc = RandomForestClassifier(max_depth=5, n_estimators=50)
    rfc.fit(X_train_res, y_train_res) 
    signature = infer_signature(X_train_res, y_train_res)

    mlflow.sklearn.log_model(
        rfc,
        "machine_failure_model_rf",
        signature=signature,
        registered_model_name="machine_failure_model_rf"
    ) 

    y_pred_train = rfc.predict(X_train)
    # Calculate the classification metrics for test data
    f1_train = f1_score(y_train, y_pred_train, average='weighted')
    accuracy_train = accuracy_score(y_train, y_pred_train)
    recall_train = recall_score(y_train, y_pred_train, average='weighted')

    # Log the classification metrics to MLflow
    mlflow.log_metric("f1_score_train", f1_train)
    mlflow.log_metric("accuracy_train", accuracy_train)
    mlflow.log_metric("recall_train", recall_train)

    # Print the run ID and the classification metrics
    print("F1 score_train:", f1_train)
    print("Accuracy_train:", accuracy_train)
    print("Recall_train:", recall_train)    

    y_pred_test = rfc.predict(X_test)
    # Calculate the classification metrics for test data
    f1_test = f1_score(y_test, y_pred_test, average='weighted')
    accuracy_test = accuracy_score(y_test, y_pred_test)
    recall_test = recall_score(y_test, y_pred_test, average='weighted')

    # Log the classification metrics to MLflow
    mlflow.log_metric("f1_score_test", f1_test)
    mlflow.log_metric("accuracy_test", accuracy_test)
    mlflow.log_metric("recall_test", recall_test)

    # Print the classification metrics
    print("F1 score_test:", f1_test)
    print("Accuracy_test:", accuracy_test)
    print("Recall_test:", recall_test)

Na saída, os conjuntos de dados de treinamento e teste geram uma pontuação F1, uma precisão e um recall de aproximadamente 0,9 ao usar o classificador de floresta aleatória.

Treinar um classificador de regressão logística

from sklearn.linear_model import LogisticRegression

with mlflow.start_run() as run:
    lr_id = run.info.run_id
    print(f"run_id {lr_id}, status: {run.info.status}")
    lr = LogisticRegression(random_state=42)
    lr.fit(X_train_res, y_train_res)
    signature = infer_signature(X_train_res, y_train_res)
  
    mlflow.sklearn.log_model(
        lr,
        "machine_failure_model_lr",
        signature=signature,
        registered_model_name="machine_failure_model_lr"
    ) 

    y_pred_train = lr.predict(X_train)
    # Calculate the classification metrics for training data
    f1_train = f1_score(y_train, y_pred_train, average='weighted')
    accuracy_train = accuracy_score(y_train, y_pred_train)
    recall_train = recall_score(y_train, y_pred_train, average='weighted')

    # Log the classification metrics to MLflow
    mlflow.log_metric("f1_score_train", f1_train)
    mlflow.log_metric("accuracy_train", accuracy_train)
    mlflow.log_metric("recall_train", recall_train)

    # Print the run ID and the classification metrics
    print("F1 score_train:", f1_train)
    print("Accuracy_train:", accuracy_train)
    print("Recall_train:", recall_train)    

    y_pred_test = lr.predict(X_test)
    # Calculate the classification metrics for test data
    f1_test = f1_score(y_test, y_pred_test, average='weighted')
    accuracy_test = accuracy_score(y_test, y_pred_test)
    recall_test = recall_score(y_test, y_pred_test, average='weighted')

    # Log the classification metrics to MLflow
    mlflow.log_metric("f1_score_test", f1_test)
    mlflow.log_metric("accuracy_test", accuracy_test)
    mlflow.log_metric("recall_test", recall_test)

Treinar um classificador XGBoost

from xgboost import XGBClassifier

with mlflow.start_run() as run:
    xgb = XGBClassifier()
    xgb_id = run.info.run_id 
    print(f"run_id {xgb_id}, status: {run.info.status}")
    xgb.fit(X_train_res.to_numpy(), y_train_res.to_numpy()) 
    signature = infer_signature(X_train_res, y_train_res)
  
    mlflow.xgboost.log_model(
        xgb,
        "machine_failure_model_xgb",
        signature=signature,
        registered_model_name="machine_failure_model_xgb"
    ) 

    y_pred_train = xgb.predict(X_train)
    # Calculate the classification metrics for training data
    f1_train = f1_score(y_train, y_pred_train, average='weighted')
    accuracy_train = accuracy_score(y_train, y_pred_train)
    recall_train = recall_score(y_train, y_pred_train, average='weighted')

    # Log the classification metrics to MLflow
    mlflow.log_metric("f1_score_train", f1_train)
    mlflow.log_metric("accuracy_train", accuracy_train)
    mlflow.log_metric("recall_train", recall_train)

    # Print the run ID and the classification metrics
    print("F1 score_train:", f1_train)
    print("Accuracy_train:", accuracy_train)
    print("Recall_train:", recall_train)    

    y_pred_test = xgb.predict(X_test)
    # Calculate the classification metrics for test data
    f1_test = f1_score(y_test, y_pred_test, average='weighted')
    accuracy_test = accuracy_score(y_test, y_pred_test)
    recall_test = recall_score(y_test, y_pred_test, average='weighted')

    # Log the classification metrics to MLflow
    mlflow.log_metric("f1_score_test", f1_test)
    mlflow.log_metric("accuracy_test", accuracy_test)
    mlflow.log_metric("recall_test", recall_test)

Etapa 5: Selecionar o melhor modelo e prever saídas

Na seção anterior, você treinou três classificadores diferentes: floresta aleatória, regressão logística e XGBoost. Você agora tem a opção de acessar programaticamente os resultados ou usar a interface do usuário (IU).

Para a opção de caminho da interface do usuário, navegue até o espaço de trabalho e filtre os modelos.

Selecione os modelos individuais para obter detalhes do desempenho do modelo.

Esse exemplo mostra como acessar programaticamente os modelos por meio do MLflow:

runs = {'random forest classifier':   rfc_id,
        'logistic regression classifier': lr_id,
        'xgboost classifier': xgb_id}

# Create an empty DataFrame to hold the metrics
df_metrics = pd.DataFrame()

# Loop through the run IDs and retrieve the metrics for each run
for run_name, run_id in runs.items():
    metrics = mlflow.get_run(run_id).data.metrics
    metrics["run_name"] = run_name
    df_metrics = df_metrics.append(metrics, ignore_index=True)

# Print the DataFrame
print(df_metrics)

Embora o XGBoost gere os melhores resultados no conjunto de treinamento, ele tem um desempenho ruim no conjunto de dados de teste. Esse desempenho ruim indica um sobreajuste. O classificador de regressão logística tem um desempenho ruim nos conjuntos de dados de treinamento e teste. No geral, o floresta aleatória atinge um bom equilíbrio entre treinar o desempenho e prevenção de sobreajuste.

Na seção a seguir, escolha o modelo floresta aleatória registrado e execute a previsão usando o recurso PREDICT:

from synapse.ml.predict import MLFlowTransformer

model = MLFlowTransformer(
    inputCols=list(X_test.columns),
    outputCol='predictions',
    modelName='machine_failure_model_rf',
    modelVersion=1
)

Com o objeto MLFlowTransformer que criou para carregar o modelo para a inferência, use a API Transformer para pontuar o modelo no conjunto de dados de teste:

predictions = model.transform(spark.createDataFrame(X_test))
predictions.show()

Esta tabela mostra a saída:

Salvar os dados no lakehouse. Os dados ficam disponíveis para usos posteriores - por exemplo, um dashboard do Power BI.

# Save test data to the lakehouse for use in the next section. 
table_name = "predictive_maintenance_test_with_predictions"
predictions.write.mode("overwrite").format("delta").save(f"Tables/{table_name}")
print(f"Spark DataFrame saved to delta table: {table_name}")

Etapa 6: Visualizar o business intelligence por meio de visualizações no Power BI

Mostrar os resultados em um formato offline, com um painel do Power BI.

O painel mostra que Tool_wear e Torque criam um limite perceptível entre os casos com e sem falha, como era esperado na análise de correlação anterior na Etapa 2.

Conteúdo relacionado

• Treinar e avaliar um modelo de classificação de texto
• Modelo de machine learning no Microsoft Fabric
• Compilar modelos de aprendizado de máquina
• Experimentos de machine learning no Microsoft Fabric