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 de Ciência de Dados do Synapse no Microsoft Fabric. O cenário usa o aprendizado de máquina para uma abordagem mais sistemática para o diagnóstico de falhas, para identificar proativamente os problemas e executar ações antes de uma falha real do computador. A meta é prever se uma máquina sofreria uma falha com base na temperatura do processo, velocidade de rotação etc.

Este tutorial aborda estas etapas:

Pré-requisitos

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

• Entre no Microsoft Fabric.

• Use o alternador de experiência no canto inferior esquerdo da sua página inicial para alternar para o Fabric.

  

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

Acompanhar em um notebook

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

• Abra e execute o notebook integrado.
• Carregue seu bloco de anotações no GitHub.

Abrir o bloco de anotações interno

O notebook de exemplo Falha de computador acompanha este tutorial.

1. Para abrir o bloco de anotações de exemplo para este tutorial, siga as instruções em Preparar seu sistema para tutoriais de ciência de dados.

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

Importar notebook do GitHub

O caderno AISample - Manutenção Preditiva acompanha este tutorial.

• Para abrir o notebook que acompanha este tutorial, siga as instruções em Preparar seu sistema para tutoriais de ciência de dados para importar o notebook para o seu workspace.

• Se você prefere copiar e colar o código desta página, pode criar um notebook.

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

Etapa 1: Instalar bibliotecas personalizadas

Para 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.

• Use as funções de instalação embutidas (%pip ou %conda) do seu notebook para instalar uma biblioteca, somente no notebook atual.
• Como alternativa, você pode criar um ambiente do Fabric, instalar bibliotecas de fontes públicas ou carregar bibliotecas personalizadas nele e, em seguida, o administrador do workspace pode anexar o ambiente como o padrão para o workspace. Todas as bibliotecas no ambiente ficarão disponíveis para uso em notebooks e na definição de tarefas do Spark no espaço de trabalho. Para obter mais informações sobre ambientes, consulte criar, configurar e usar um ambiente no Microsoft Fabric.

Para este tutorial, use %pip install para instalar a biblioteca de imblearn em seu 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 em configurações industriais. Ele 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 10000

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

• Temperatura do ar, em graus Kelvin (K)

• Temperatura do processo, em graus Kelvin

• Velocidade rotacional, 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, respectivamente, 5, 3 e 2 minutos de desgaste à ferramenta usada no processo.

• Um Rótulo de Falha do Computador, para indicar se o computador 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:

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

Baixar o conjunto de dados e carregar no lakehouse

Conecte-se ao contêiner de Conjuntos de Dados Abertos do Azure e carregue o conjunto de dados de Manutenção Preditiva. Esse código baixa uma versão disponível publicamente do conjunto de dados e, em seguida, armazena-a 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 de baixar o conjunto de dados no lakehouse, você pode 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 visualização dos dados:

Gravar um DataFrame do Spark em uma tabela Delta do lakehouse

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

# 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 colunas 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 de dados exploratória

Converta o DataFrame do Spark em um DataFrame do Pandas para utilizar bibliotecas populares de plotagem compatíveis com 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 colunas específicas do conjunto de dados em floats ou tipos inteiros conforme necessário e mapeie cadeias de caracteres ('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 o esperado, a falha (IsFail) correlaciona-se com os recursos selecionados (colunas). A matriz de correlação mostra que Air_temperature, Process_temperature, Rotational_speed, Torquee 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, Torquee Tool_wear não são esparsas. Eles parecem ter uma boa continuidade no espaço do recurso. Esses gráficos confirmam que treinar um modelo de machine learning nesse conjunto de dados provavelmente produz resultados confiáveis que podem generalizar para um novo conjunto de dados.

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

Conte o número de amostras para computadores com falha e sem falha e inspecione o equilíbrio de dados de 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 dos exemplos. Utilize 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 a classe minoritária tem poucos exemplos para que o modelo aprenda efetivamente o limite de decisão.

SMOTE pode resolver o problema. SMOTE é uma técnica de superamostragem amplamente usada que gera exemplos sintéticos. Ele gera exemplos para a classe minoritária com base nas distâncias euclidianas entre pontos de dados. Esse método difere da sobrecarga aleatória, pois 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 desequilibrado.

# 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ê balanceou com êxito o conjunto de dados. Agora você pode migrar para o treinamento de modelo.

Etapa 4: Treinar e avaliar os modelos

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

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

Treinar um 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 apresentam uma medida f, precisão e recall de cerca de 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. Agora você tem a opção de acessar programaticamente os resultados ou usar a interface do usuário (interface do usuário).

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

Selecione modelos individuais para obter detalhes do desempenho do modelo.

Este 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 produza os melhores resultados no conjunto de treinamento, ele tem um desempenho ruim no conjunto de dados de teste. Esse desempenho ruim indica sobreajuste. O classificador de regressão logística tem um desempenho ruim em conjuntos de dados de treinamento e teste. No geral, a floresta aleatória atinge um bom equilíbrio entre desempenho de treino e prevenção do sobreajuste.

Na próxima seção, escolha o modelo de floresta aleatória registrado e execute uma previsão com 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 você criou para carregar o modelo para 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:

Salve os dados no lakehouse. Os dados ficam disponíveis para usos posteriores, por exemplo, um painel 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: exibir 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 de controle mostra que Tool_wear e Torque criam um limite perceptível entre casos com falha e sem falha, como esperado a partir da análise de correlação anterior no passo 2.

Conteúdo relacionado

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