Kurz: vytvorenie, vyhodnotenie a skóre modelu zisťovania chýb počítača

Tento kurz predstavuje komplexný príklad pracovného postupu synapse Data Science v službe Microsoft Fabric. Tento scenár používa strojové učenie na systematickejší prístup k diagnostike chýb, proaktívne identifikuje problémy a podniká kroky pred skutočným zlyhaním počítača. Cieľom je predpovedať, či sa v počítači vyskytne zlyhanie na základe teploty procesu, rýchlosti otáčania atď.

V tomto kurze sú obsiahnuté tieto kroky:

Predpoklady

• Získajte predplatné služby Microsoft Fabric . Alebo si zaregistrujte bezplatnú skúšobnú služby Microsoft Fabric.

• Prihláste sa do služby Microsoft Fabric.

• Pomocou prepínača skúseností v ľavej dolnej časti domovskej stránky sa prepnete na službu Fabric.

  

• Ak je to potrebné, vytvorte si podľa popisu v téme Vytvorenie jazera v službe Microsoft Fabric.

Sledovanie v poznámkovom bloke

Môžete si vybrať jednu z týchto možností, ktorú si môžete vyskúšať v notebooku:

• Otvorte a spustite vstavaný poznámkový blok.
• Nahrajte poznámkový blok z GitHubu.

Otvorenie vstavaného poznámkového bloku

K tomuto kurzu sa dopĺňa vzorový poznámkový blok zlyhaní počítača počítači.

1. Ak chcete otvoriť vzorový poznámkový blok pre tento kurz, postupujte podľa pokynov v téme Príprava systému na kurzy dátovej vedy.

2. Uistite sa, že pripojiť lakehouse na notebook, ako začnete bežať kód.

Importovanie notebooku z GitHubu

Tento kurz sprevádza poznámkový blok AISample – prediktívna údržba.

• Ak chcete otvoriť sprievodný poznámkový blok pre tento kurz, postupujte podľa pokynov v téme Príprava systému na kurzy dátových vied na import notebooku do pracovného priestoru.

• Ak by ste radšej skopírovali a prilepili kód z tejto stránky, môžete vytvoriť nový poznámkový blok.

• Uistite sa, že pripojiť lakehouse k notebooku, ako začnete bežať kód.

Krok č. 1: Inštalácia vlastných knižníc

Na vývoj modelu strojového učenia alebo analýzu údajov ad hoc možno budete musieť rýchlo nainštalovať vlastnú knižnicu pre reláciu Apache Spark. Na inštaláciu knižníc máte dve možnosti.

• Ak chcete nainštalovať knižnicu iba v aktuálnom poznámkovom bloku, použite možnosti vnorenej inštalácie (%pip alebo %conda).
• Prípadne môžete vytvoriť prostredie služby Fabric, nainštalovať knižnice z verejných zdrojov alebo do neho nahrať vlastné knižnice a potom môže správca pracovného priestoru k prostrediu pripojiť ako predvolené prostredie pre pracovný priestor. Všetky knižnice v prostredí budú potom k dispozícii na použitie v poznámkových blokoch a definíciách úloh služby Spark v pracovnom priestore. Ďalšie informácie o prostrediach nájdete v téme vytvorenia, konfigurácie a používania prostredia v službe Microsoft Fabric.

V tomto kurze použite %pip install na inštaláciu knižnice imblearn v poznámkovom bloku.

# Use pip to install imblearn
%pip install imblearn

Krok č. 2: Načítanie údajov

Množina údajov simuluje zapisovanie parametrov výrobného stroja do denníka ako funkciu času, ktorá je bežná v priemyselných nastaveniach. Skladá sa z 10 000 údajových bodov uložených ako riadky s funkciami ako stĺpce. K funkciám patria:

• Jedinečný identifikátor (UID), ktorý je v rozsahu od 1 do 1 000

• ID produktu pozostávajúce z písmena L (pre nízku hodnotu), M (pre stredné) alebo H (s vysokým) na označenie variantu kvality produktu a sériového čísla špecifického variantu. Varianty nízkej, strednej a vysokej kvality tvoria 60%, 30%a 10% všetkých produktov.

• Teplota vzduchu v stupňoch Kelvin (K)

• Teplota procesu v stupňoch Kelvin

• Otočná rýchlosť v revolúciách za minútu (RPM)

• Moment, v Newton-Meters (Nm)

• Opotrebenie nástroja v minútach. Kvalitné varianty H, M, a L pridajú 5, 3 a 2 minúty opotrebenia nástroja do nástroja použitého v procese

• Označenie zlyhania počítača na označenie toho, či počítač zlyhal v konkrétnom údajovom bode. Tento konkrétny údajový bod môže mať ktorýkoľvek z nasledujúcich piatich nezávislých režimov zlyhania:

  • Zlyhanie opotrebenia nástroja (TWF): nástroj sa nahradí alebo zlyhá pri náhodne vybratom čase opotrebenia 200 až 240 minút.
  • Zlyhanie rozplynutia tepla (HDF): Rozpustenie tepla spôsobí zlyhanie procesu, ak je rozdiel medzi teplotou vzduchu a teplotou procesu menší ako 8,6 K a otočná rýchlosť nástroja je menšia ako 1380 RPM
  • Zlyhanie prúdu (PWF): Súčin momentu a rýchlosti rotácie (v radoch) sa rovná výkonu požadovanému pre proces. Proces zlyhá, ak výkon klesne pod hodnotu 3 500 W alebo prekročí hodnotu 9 000 W.
  • Zlyhanie preťaženia (OSF): ak súčin opotrebenia a krútiaceho momentu prekročí 11 000 minimálnych nm pre variant produktu L (12 000 pre M, 13 000 pri H), proces zlyhá v dôsledku preťaženia
  • Náhodné zlyhania (RNF): každý proces má šancu na zlyhanie 0,1%, bez ohľadu na parametre procesu

Stiahnite si množinu údajov a nahrajte ju do služby lakehouse

Pripojte sa ku kontajneru Azure Open Datasets a načítajte množinu údajov Prediktívna údržba. Tento kód stiahne verejne dostupnú verziu množiny údajov a potom ju uloží v úložisku Fabric lakehouse:

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

Po stiahnutí množiny údajov do úložiska lakehouse ju môžete načítať ako údajový rámec Spark:

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

V tejto tabuľke sa zobrazuje ukážka údajov:

Zápis prvku Spark DataFrame do delta tabuľky lakehouse

Naformátujte údaje (napríklad nahraďte medzery znakom podčiarknutia) a v nasledujúcich krokoch uľahčíte operácie služby Spark:

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

Táto tabuľka zobrazuje ukážku údajov s preformátovanými názvami stĺpcov:

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

Krok č. 3: Predbežné vypracovanie údajov a vykonanie prieskumných analýz údajov

Konvertujte údajový rámec Spark na údajový rámec Pandas, aby ste mohli používať obľúbené knižnice na vykresľovanie kompatibilné s knižnicou 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()

Konvertujte podľa potreby konkrétne stĺpce množiny údajov na typy pohyblivých čísel alebo celé čísla a reťazce mapy ('L', 'M', 'H') na číselné hodnoty (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})

Skúmanie údajov prostredníctvom vizualizácií

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

Podľa očakávania zlyhanie (IsFail) koreluje s vybratými funkciami (stĺpcami). Matica korelácie ukazuje, že Air_temperature, Process_temperature, Rotational_speed, Torquea Tool_wear majú najvyššiu koreláciu s premennou 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])

Ako zobrazujú vykreslené grafy, premenné Air_temperature, Process_temperature, Rotational_speed, Torquea Tool_wear nie sú riedke. Zdá sa, že majú dobrú kontinuitu v priestore funkcie. Tieto vykreslenia potvrdzujú, že trénovanie modelu strojového učenia na tejto množine údajov pravdepodobne prinesie spoľahlivé výsledky, ktoré možno zovšeobecniť na novú množinu údajov.

Skontrolujte cieľovú premennú, či sa nevyváži trieda

Spočítajte počet vzoriek pre neúspešné a neobslúžené stroje a skontrolujte zostatok údajov každej triedy (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()

Na vykresleniach je uvedené, že trieda bez zlyhania (zobrazená ako IsFail=0 v druhom vykreslenia) predstavuje väčšinu vzoriek. Pomocou techniky prekročenia objemu údajov vytvorte vyváženejšiu tréningovú množinu údajov:

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

Prekrývanie a vyváženie tried v tréningovej množine údajov

Predchádzajúca analýza ukázala, že množina údajov je veľmi nevyvážená. Táto nerovnováha sa stáva problémom, pretože menšinová trieda má príliš málo príkladov na to, aby model účinne naučil rozhodovaciu hranicu.

SMOTE môže vyriešiť tento problém. SMOTE je široko používaná technika prekvapnutia, ktorá generuje syntetické príklady. Generuje príklady menšinovej triedy založenej na euklidiánskych vzdialenostiach medzi údajovými bodmi. Táto metóda sa líši od náhodného prekvapovania, pretože vytvára nové príklady, ktoré nezduplikujú len menšinovú triedu. Metóda sa stáva efektívnejsiehodnejším spôsobom na zvládnutie nevyvážených množín údajov.

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

Úspešne vyvážená množina údajov. Teraz môžete prejsť na trénovanie modelu.

Krok č. 4: Trénovanie a vyhodnotenie modelov

MLflow zaregistruje modely, vlaky a porovnáva rôzne modely a vyberá najlepší model na účely predpovede. Na trénovanie modelu môžete použiť nasledujúce tri modely:

• Náhodný klasifikant lesa
• Klasifikant logistickej regresie
• Klasifikant XGBoost

Trénovať náhodný klasifikant lesa

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)

Z výstupu majú tréningové aj testovacie množiny údajov výnos skóre F1, presnosť a úplnosť hodnoty asi 0,9 pri použití náhodného klasifikátu lesa.

Trénovanie logistickej regresnej klasifikácie

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)

Trénovanie klasifikanta 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)

Krok č. 5: Vyberte najlepší model a predpovedať výstupy

V predchádzajúcej časti ste trénovali tri rôzne klasifikácie: náhodnú lesnú, logistickú regresiu a XGBoost. Teraz máte možnosť buď programovo získať prístup k výsledkom, alebo použiť používateľské rozhranie (UI).

V prípade možnosti cesty používateľského rozhrania prejdite do svojho pracovného priestoru a filtrujte modely.

Podrobnosti o výkone modelu získate výberom jednotlivých modelov.

Tento príklad ukazuje, ako programovo získať prístup k modelom prostredníctvom toku 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)

Hoci XGBoost prináša najlepšie výsledky v tréningovej množine, na testovacej množine údajov funguje zle. Tento slabý výkon označuje prekročenie výkonu. Klasifikant logistickej regresie má slabý výkon na trénovaní aj testovaní množín údajov. Náhodný les vytvára dobrú rovnováhu medzi výkonom školenia a vyhýbaním sa nadmernému započítavaniu.

V ďalšej časti vyberte registrovaný model náhodného lesa a vykonajte predpoveď s funkciou PREDICT:

from synapse.ml.predict import MLFlowTransformer

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

S objektom MLFlowTransformer, ktorý ste vytvorili na načítanie modelu na odvodenie, použite rozhranie API transformátora na skóre modelu v testovacej množine údajov:

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

V tejto tabuľke je zobrazený výstup:

Uložte údaje do útla Lakehouse. Údaje budú k dispozícii na neskoršie použitie, napríklad na tabuľu služby 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}")

Krok č. 6: Zobrazenie Business Intelligence prostredníctvom vizualizácií v Power BI

Zobrazte výsledky v offline formáte s tabuľou Power BI.

Tabuľa zobrazuje, že Tool_wear a Torque vytvoriť viditeľnú hranicu medzi neúspešnými a nevyliečenými prípadmi, ako sa očakáva z predchádzajúcej korelačnej analýzy v kroku 2.

Súvisiaci obsah

• Trénovať a vyhodnotiť model textovej klasifikácie
• modelu strojového učenia v službe Microsoft Fabric
• modely strojového učenia
• experimentov strojového učenia v službe Microsoft Fabric