教程：创建、评估机器故障检测模型并对其评分

本教程介绍了 Microsoft Fabric 中 Synapse 数据科学工作流的端到端示例。 该方案旨在利用机器学习获得更系统的故障诊断方法，以便在机器发生实际故障之前主动识别问题并采取行动。 目标是根据过程温度、转速等特征来预测机器是否会出现故障。

本教程涵盖以下步骤：

先决条件

• 获取 Microsoft Fabric 订阅。 或者注册免费的 Microsoft Fabric 试用版。

• 登录 Microsoft Fabric。

• 使用主页左侧的体验切换器切换到 Synapse 数据科学体验。

  

• 如有必要，请如在 Microsoft Fabric 中创建湖屋中所述创建 Microsoft Fabric 湖屋。

请按照笔记本进行操作

可以选择下面其中一个选项，以在笔记本中进行操作：

• 在数据科学体验中打开并运行内置笔记本
• 将笔记本从 GitHub 上传到数据科学体验

打开内置笔记本

“计算机故障”是本教程随附的示例笔记本。

要在 Synapse 数据科学体验中打开教程的内置示例笔记本，请执行以下操作：

1. 转到 Synapse 数据科学主页。

2. 选择“使用示例”。

3. 选择相应的示例：

   • 来自默认的“端到端工作流 (Python)”选项卡（如果示例适用于 Python 教程）。
   • 来自“端到端工作流 (R)“选项卡（如果示例适用于 R 教程）。
   • 从“快速教程”选项卡选择（如果示例适用于快速教程）。

4. 在开始运行代码之前，将湖屋连接到笔记本。

从 GitHub 导入笔记本

“AISample – 预测性维护”是本教程随附的笔记本。

若要打开本教程随附的笔记本，请按照让系统为数据科学做好准备教程中的说明操作，将该笔记本导入到工作区。

或者，如果要从此页面复制并粘贴代码，则可以创建新的笔记本。

在开始运行代码之前，请务必将湖屋连接到笔记本。

步骤 1：安装自定义库

对于机器学习模型开发或临时数据分析，可能需要为 Apache Spark 会话快速安装自定义库。 有两个选项可用于安装库。

• 使用笔记本的内联安装功能（%pip 或 %conda），仅在当前笔记本中安装库。
• 也可以创建 Fabric 环境，安装来自公共来源的安装库或将自定义库上传到该环境，然后工作区管理员可以将环境附加为工作区的默认值。 然后，环境中的所有库都将可用于工作区中的任何笔记本和 Spark 作业定义。 有关环境的详细信息，请参阅在 Microsoft Fabric 中创建、配置和使用环境。

在本教程中，使用 %pip install 在笔记本中安装 imblearn 库。

# Use pip to install imblearn
%pip install imblearn

步骤 2：加载数据

数据集模拟将制造机器的参数记录为时间函数，这在工业环境中很常见。 它由 10,000 个数据点组成，这些数据点存储为行，特征存储为列。 具体功能包括：

• 唯一标识符 (UID)，范围从 1 到 10000

• 产品 ID，由字母 L（低）、M（中）或 H（高）组成，表示产品质量变体以及特定于变体的序列号。 低质量、中质量和高质量变体分别占所有产品的 60%、30% 和 10%

• 空气温度，以开尔文 (K) 为单位

• 过程温度，以开尔文为单位

• 转速，以转/分钟 (RPM) 为单位

• 扭矩，以牛顿米 (Nm) 为单位

• 工具磨损时间，以分钟为单位。 质量变体 H、M 和 L 分别为过程中使用的工具增加 5 分钟、3 分钟和 2 分钟的工具磨损时间。

• 计算机故障标签，用于指示计算机是否在特定数据点发生故障。 此特定数据点可具有以下五种独立故障模式中的任何一种：

  • 工具磨损故障 (TWF)：工具在随机选择的工具磨损时间（200 到 240 分钟）更换或发生故障
  • 散热故障 (HDF)：如果空气和古城温度之间的差异低于 8.6 K 并且工具的转速低于 1380 RPM，则散热会导致过程故障
  • 电源故障 (PWF)：扭矩和转速（以 rad/s 为单位）的乘积等于过程所需的功率。 如果该功率低于 3500 W 或高于 9000 W，则该过程将失败
  • 过度应变故障 (OSF)：如果 L 产品变体（12,000 M，13,000 H）的工具磨损时间和扭矩的乘积超过 11,000 分钟牛米，则该过程会因过度应变而失败
  • 随机故障 (RNF)：每个过程都有 0.1% 的几率发生故障，无论其过程参数如何

下载数据集并上传到湖屋

连接到 Azure 开放数据集容器并加载预测性维护数据集。 以下代码将下载数据集的公开可用版本，然后将其存储在 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.")

将数据集下载到湖屋后，可以将其加载为 Spark DataFrame：

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

下表显示了数据的预览：

将 Spark DataFrame 写入湖屋增量表

设置数据格式（例如用下划线替换空格），方便后续步骤的 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)

下表显示了具有重格式化列名称的数据的预览：

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

步骤 3：预处理数据并执行探索性数据分析

将 Spark 数据帧转换为 Pandas 数据帧，以使用与 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()

将数据集的特定列转换为所需的浮点数或整数类型，并将字符串（'L'、'M'、'H'）映射为数值（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})

通过可视化效果浏览数据

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

正如预期的那样，失败 (IsFail) 与所选特征（列）相关。 相关矩阵显示 Air_temperature、Process_temperature、Rotational_speed、Torque 和 Tool_wear 与 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])

如绘制图表中所示，Air_temperature、Process_temperature、Rotational_speed、Torque 和 Tool_wear 变量并不稀疏。 它们似乎在特征空间中表现出良好的连续性。 这些绘图证实，在此数据集上训练机器学习模型可能会产生可靠的结果，可以推广到新数据集。

检查目标变量的类别不平衡

计算发生故障和未发生故障机器的样本数量，并检查每个类别的数据平衡（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()

这些绘图表明无故障类别（在第二个绘图中显示为 IsFail=0）构成了大多数样本。 使用过采样技术创建更平衡的训练数据集：

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

过采样以平衡训练数据集中的类别

前面的分析表明数据集高度不平衡。 不平衡成为问题的原因在于，少数类的示例太少，模型无法有效了解决策边界。

SMOTE 可以解决问题。 SMOTE 这是一种广泛使用的过采样技术，可生成合成示例。 它以欧几里得距离为基础，根据数据点之间的距离生成少数类的示例。 此方法不同于随机过度采样，因为它创建的新示例不只复制少数类。 该方法可更有效地处理不平衡数据集。

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

你已成功平衡数据集。 现在可以开始模型训练。

步骤 4：训练和评估模型

MLflow 用于注册模型、训练和比较各种模型，以及选取最佳模型进行预测。 可使用以下 3 个模型进行模型训练：

• 随机林分类器
• 逻辑回归分类器
• XGBoost 分类器

训练随机林分类器

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)

从输出来看，使用随机林分类器时，训练数据集和测试数据集都会产生约为 0.9 的 F1 分数、准确度和召回率。

训练逻辑回归分类器

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)

训练 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)

步骤 5：选择最佳模型并预测输出

在上一部分，你训练了三种不同的分类器：随机林、逻辑回归和 XGBoost。 现在可以选择以编程方式访问结果，也可以使用用户界面 (UI)。

若要选择 UI 路径，请导航到工作区并筛选模型。

选择各个模型以了解模型性能的详细信息。

该示例演示了如何通过 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)

尽管 XGBoost 在训练集上获得了最好的结果，但它在测试数据集上的表现很差，这表明存在过度拟合。 性能不佳表示过度拟合。 逻辑回归分类器在训练和测试数据集上都表现不佳。 总体而言，随机林在训练性能和避免过度拟合之间取得了良好的平衡。

在下一部分中，选择已注册的随机林模型并使用 PREDICT 功能执行预测：

from synapse.ml.predict import MLFlowTransformer

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

使用创建的 MLFlowTransformer 对象来加载用于推理的模型，然后使用 Transformer API 在测试数据集上对模型进行评分：

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

此表显示输出：

将数据保存到湖屋中。 然后，数据可供以后使用，例如用于 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}")

步骤 6：在 Power BI 中通过可视化效果查看商业智能

使用 Power BI 仪表板以脱机格式显示结果。

仪表板显示发生故障和未发生故障案例的 Tool_wear 和 Torque 之间存在明显区别，正如步骤 2 中早期相关分析所预期的那样。

相关内容

• 训练和评估文本分类模型
• Microsoft Fabric 中的机器学习模型
• 训练机器学习模型
• Microsoft Fabric 中的机器学习试验