Esercitazione: creare, valutare e assegnare un punteggio a un modello di rilevamento degli errori del computer

Questa esercitazione presenta un esempio end-to-end di un flusso di lavoro di data science Synapse in Microsoft Fabric. Lo scenario usa l'apprendimento automatico per un approccio più sistematico alla diagnosi degli errori, per identificare in modo proattivo i problemi e intraprendere azioni prima di un errore effettivo del computer. L'obiettivo è prevedere se un computer riscontra un errore in base alla temperatura del processo, alla velocità rotazionale e così via.

Questa esercitazione comprende i seguenti passaggi:

Prerequisiti

• Ottieni una sottoscrizione a Microsoft Fabric. In alternativa, iscriversi per una versione gratuita di valutazione di Microsoft Fabric.

• Accedi a Microsoft Fabric.

• Passa a Fabric usando il commutatore nell'angolo in basso a sinistra della tua home page.

  

• Se necessario, crea un lakehouse di Microsoft Fabric come descritto in "Creare un lakehouse in Microsoft Fabric".

Seguire e annotare in un quaderno

È possibile scegliere una di queste opzioni per seguire la procedura in un notebook:

• Aprire ed eseguire il notebook integrato.
• Carica il tuo notebook da GitHub.

Aprire il notebook predefinito

Il notebook di esempio Guasto macchina accompagna questa esercitazione.

1. Per aprire il notebook di esempio per questa esercitazione, seguire le istruzioni riportate in Preparare il sistema per le esercitazioni sull'analisi scientifica dei dati.

2. Assicurati di collegare una lakehouse al notebook prima di eseguire il codice.

Importare il notebook da GitHub

Il notebook AISample - Manutenzione Predittiva accompagna questo tutorial.

• Per aprire il notebook di accompagnamento per questa esercitazione, seguire le istruzioni riportate in Preparare il sistema per le esercitazioni di data science per importare il notebook nell'area di lavoro.

• Se preferisci copiare e incollare il codice da questa pagina, puoi creare un nuovo notebook.

• Assicurati di collegare un lakehouse al notebook prima di iniziare a eseguire il codice.

Passaggio 1: Installare librerie personalizzate

Per lo sviluppo di modelli di apprendimento automatico o l'analisi dei dati ad hoc, potrebbe essere necessario installare rapidamente una libreria personalizzata per la sessione di Apache Spark. Sono disponibili due opzioni per installare le librerie.

• Usare le funzionalità di installazione inline (%pip o %conda) del notebook per installare una libreria solo nel notebook corrente.
• In alternativa, è possibile creare un ambiente Fabric, installare librerie da origini pubbliche o caricarvi librerie personalizzate e quindi l'amministratore dell'area di lavoro può collegare l'ambiente come predefinito per l'area di lavoro. Tutte le librerie in questo ambiente diventeranno quindi disponibili per l'uso in qualsiasi notebook e in tutte le definizioni di job Spark nell'area di lavoro. Per altre informazioni sugli ambienti, vedere creare, configurare e usare un ambiente in Microsoft Fabric.

Per questa esercitazione, usare %pip install per installare la libreria imblearn nel notebook.

# Use pip to install imblearn
%pip install imblearn

Passaggio 2: caricare i dati

Il set di dati simula la registrazione dei parametri di una macchina di produzione come funzione del tempo, che è comune nell'industria. È costituito da 10.000 punti dati archiviati come righe con funzionalità come colonne. Queste funzionalità comprendono:

• Identificatore univoco (UID) compreso tra 1 e 10000

• ID prodotto, costituito da una lettera L (per basso), M (per medio) o H (per alto), per indicare la variante di qualità del prodotto e un numero di serie specifico della variante. Le varianti di bassa, media e alta qualità costituiscono rispettivamente il 60%, il 30% e il 10% di tutti i prodotti

• Temperatura dell'aria, in gradi Kelvin (K)

• La temperatura di processo, in gradi kelvin

• Velocità rotazionale, in rivoluzioni al minuto (RPM)

• Coppia, in Newton-Metri (Nm)

• Usura degli strumenti in minuti. Le varianti di qualità H, M e L aggiungono rispettivamente 5, 3 e 2 minuti di usura dello strumento usato nel processo

• Etichetta guasto macchina per indicare se la macchina ha avuto un guasto nel punto dati specifico. Questo punto dati specifico può avere una delle cinque modalità di errore indipendenti seguenti:

  • Errore di usura degli strumenti (TWF): lo strumento viene sostituito o ha esito negativo in un tempo di usura casuale degli strumenti selezionato, tra 200 e 240 minuti
  • Errore di dissipazione termica (HDF): la dissipazione termica causa un errore di processo se la differenza tra la temperatura dell'aria e la temperatura del processo è inferiore a 8,6 K e la velocità rotazionale dello strumento è inferiore a 1380 RPM
  • Guasto elettrico (PWF): il prodotto della coppia e della velocità rotazionale (in rad/s) è uguale alla potenza necessaria per il processo. Il processo ha esito negativo se questa potenza scende al di sotto di 3.500 W o supera 9.000 W
  • OverStrain Failure (OSF): se il prodotto di usura degli strumenti e coppia supera 11.000 Nm per la variante di prodotto L (12.000 per M, 13.000 per H), il processo fallisce a causa di sovraccarico.
  • Errori casuali (RNF): ogni processo ha una probabilità di errore pari a 0,1%, indipendentemente dai parametri del processo

Scaricare il set di dati e caricarlo nel lakehouse

Connettersi al contenitore Azure Open Datasets e caricare il set di dati di manutenzione predittiva. Questo codice scarica una versione disponibile pubblicamente del set di dati, quindi la archivia in un lakehouse di 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.z13.web.core.windows.net/data/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.")

Dopo aver scaricato il set di dati nel lakehouse, è possibile caricarlo come dataframe Spark:

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

Questa tabella mostra un’anteprima dei dati:

Scrivere un DataFrame Spark in una tabella Delta Lakehouse

Formattare i dati (ad esempio, sostituire gli spazi con caratteri di sottolineatura) per facilitare le operazioni Spark nei passaggi successivi:

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

Questa tabella mostra un'anteprima dei dati con nomi di colonna riformattati:

# 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}")

Passaggio 3: Pre-elaborare i dati ed eseguire l'analisi esplorativa dei dati

Convertire il dataframe Spark in un dataframe pandas per usare librerie di tracciate popolari compatibili con 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()

Convertire colonne specifiche del set di dati in tipi float o integer in base alle esigenze e mappare le stringhe ('L', 'M', 'H') in valori numerici (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})

Esplorare i dati tramite visualizzazioni

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

Come previsto, il fallimento (IsFail) è correlato alle caratteristiche selezionate (colonne). La matrice di correlazione mostra che Air_temperature, Process_temperature, Rotational_speed, Torque, e Tool_wear hanno la correlazione più alta con la variabile 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])

Come mostrano i grafici tracciati, le variabili Air_temperature, Process_temperature, Rotational_speed, Torque e Tool_wear non sono sparse. Sembrano avere una buona continuità nello spazio delle funzionalità. Questi tracciati confermano che il training di un modello di Machine Learning in questo set di dati produce probabilmente risultati affidabili che possono generalizzare in un nuovo set di dati.

Esaminare la variabile di destinazione per individuare lo squilibrio tra le classi

Contare il numero di campioni per le macchine non riuscite e riuscite ed esaminare il bilanciamento dei dati per ogni 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()

I tracciati indicano che la classe senza errori (mostrata come IsFail=0 nel secondo tracciato) rappresenta la maggior parte dei campioni. Utilizzare una tecnica di sovracampionamento per creare un set di dati di addestramento più bilanciato.

# 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}")

Oversampling per bilanciare le classi nel set di dati di addestramento

L'analisi precedente ha mostrato che il set di dati è altamente sbilanciato. Questo squilibrio diventa un problema, perché la classe di minoranza ha troppo pochi esempi affinché il modello possa apprendere efficacemente il limite decisionale.

SMOTE può risolvere il problema. SMOTE è una tecnica di sovracampionamento ampiamente usata che genera esempi sintetici. Genera esempi per la classe di minoranza in base alle distanze euclidee tra i punti dati. Questo metodo è diverso dal sovracampionamento casuale, perché crea nuovi esempi che non duplicano solo la classe di minoranza. Il metodo diventa una tecnica più efficace per gestire set di dati sbilanciati.

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

Il set di dati è stato bilanciato correttamente. È ora possibile passare al training del modello.

Passaggio 4: Eseguire il training e valutare i modelli

MLflow registra modelli, esegue il training e confronta vari modelli, e seleziona il modello migliore a fini di previsione. È possibile usare i tre modelli seguenti per il training del modello:

• Classificatore tramite foresta casuale
• Classificatore di regressione logistica
• Classificatore XGBoost

Eseguire il training di un classificatore di foresta casuale

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)

Dall'output, sia il set di dati di addestramento che quello di test producono un punteggio F1, un'accuratezza e un recall di circa 0,9 quando si utilizza il classificatore random forest.

Addestrare un classificatore di regressione logistica

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)

Addestrare un classificatore 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)

Passaggio 5: Selezionare il modello migliore e prevedere gli output

Nella sezione precedente sono stati sottoposti a training tre classificatori diversi: foresta casuale, regressione logistica e XGBoost. È ora possibile scegliere di accedere ai risultati a livello di codice o di usare l'interfaccia utente.

Per l'opzione percorso UI, vai all'area di lavoro e filtra i modelli.

Selezionare singoli modelli per informazioni dettagliate sulle prestazioni del modello.

Questo esempio illustra come accedere ai modelli a livello di codice tramite 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)

Anche se XGBoost restituisce i migliori risultati nel set di training, ha una scarsa prestazione sul set di test. Le scarse prestazioni indicano un sovradattamento. Il classificatore di regressione logistica ha prestazioni scarse nei set di dati di training e di test. Nel complesso, il random forest raggiunge un buon equilibrio tra le prestazioni di addestramento e l'evitare l'overadattamento.

Nella sezione successiva, scegliere il modello di foresta random registrato ed eseguire una previsione con la funzionalità PREDICT:

from synapse.ml.predict import MLFlowTransformer

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

Con l'oggetto MLFlowTransformer creato per caricare il modello per l'inferenza, usa l'API Transformer per assegnare un punteggio al modello sul set di dati di test.

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

Questa tabella mostra l'output:

Salvare i dati nel sistema lakehouse. I dati diventano quindi disponibili per usi successivi, ad esempio un dashboard di 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}")

Passaggio 6: Visualizzare business intelligence tramite visualizzazioni in Power BI

Visualizzare i risultati in un formato offline, con un dashboard di Power BI.

Il dashboard mostra che Tool_wear e Torque creano un limite evidente tra i casi falliti e non falliti, come previsto dall'analisi di correlazione precedente nel passaggio 2.

Contenuto correlato

• Eseguire il training e la valutazione di un modello di classificazione di testo
• Modello di machine learning in Microsoft Fabric
• Allenare modelli di apprendimento automatico
• Esperimenti di apprendimento automatico in Microsoft Fabric