使用Amazon SageMaker Jumpstart预测车队故障概率
Predicting fleet failure probability using Amazon SageMaker Jumpstart.
预测性维护在汽车行业中至关重要,因为它可以避免突发机械故障和破坏运营的反应性维护活动。通过预测车辆故障并安排维护和修理,您将减少停机时间,提高安全性,并提高生产力水平。
如果我们能够将深度学习技术应用于驱动车辆故障、计划外停机和维修成本的常见领域,会怎样呢?
在本文中,我们将展示如何使用Amazon SageMaker JumpStart训练和部署一个模型来预测车辆故障概率。SageMaker Jumpstart是Amazon SageMaker的机器学习(ML)中心,提供预训练的、公开可用的模型,适用于各种问题类型,帮助您开始进行机器学习。本文中概述的解决方案可在GitHub上找到。
SageMaker JumpStart解决方案模板
SageMaker JumpStart为许多常见的ML用例提供一键式的端到端解决方案。请浏览以下用例,以获取有关可用解决方案模板的更多信息:
- 需求预测
- 信用评级预测
- 欺诈检测
- 计算机视觉
- 从文档中提取和分析数据
- 预测性维护
- 流失预测
- 个性化推荐
- 强化学习
- 医疗保健和生命科学
- 金融定价
SageMaker JumpStart解决方案模板涵盖各种用例,每个用例下都提供了几个不同的解决方案模板(本文中的解决方案“车队预测性维护”位于“解决方案”部分)。请选择最适合您用例的解决方案模板,从SageMaker JumpStart入口页面开始。有关每个用例下特定解决方案的更多信息以及如何启动SageMaker JumpStart解决方案,请参阅解决方案模板。
解决方案概述
AWS针对汽车车队的预测性维护解决方案应用了深度学习技术,用于处理驱动车辆故障、计划外停机和维修成本的常见领域。它作为一个初始的基础模块,帮助您在短时间内得到一个概念验证。该解决方案在SageMaker中包含数据准备和可视化功能,允许您对深度学习模型的超参数进行训练和优化。您可以使用自己的数据或作为解决方案一部分使用合成数据集。该版本处理随时间变化的车辆传感器数据。后续版本将处理维护记录数据。
下图展示了您如何使用SageMaker组件来使用该解决方案。作为解决方案的一部分,使用了以下服务:
- Amazon S3 – 我们使用Amazon Simple Storage Service(Amazon S3)存储数据集
- SageMaker笔记本 – 我们使用笔记本来预处理和可视化数据,并训练深度学习模型
- SageMaker端点 – 我们使用端点部署训练的模型
工作流程包括以下步骤:
- 从车队管理系统中创建包含车辆数据和传感器日志的历史数据提取。
- 训练ML模型后,部署SageMaker模型工件。
- 连接的车辆通过AWS IoT Core发送传感器日志(或通过HTTP接口发送)。
- 传感器日志通过Amazon Kinesis Data Firehose持久化。
- 传感器日志发送到AWS Lambda以针对模型进行查询并进行预测。
- Lambda将传感器日志发送到Sagemaker模型推断进行预测。
- 预测结果持久化在Amazon Aurora中。
- 聚合结果在Amazon QuickSight仪表板上显示。
- 关于故障预测概率的实时通知发送到Amazon Simple Notification Service(Amazon SNS)。
- Amazon SNS将通知发送回连接的车辆。
该解决方案包含六个笔记本:
- 0_demo.ipynb – 我们解决方案的快速预览
- 1_introduction.ipynb – 介绍和解决方案概述
- 2_data_preparation.ipynb – 准备示例数据集
- 3_data_visualization.ipynb – 可视化我们的示例数据集
- 4_model_training.ipynb – 在我们的示例数据集上训练一个用于检测故障的模型
- 5_results_analysis.ipynb – 分析我们训练的模型的结果
先决条件
Amazon SageMaker Studio是SageMaker内的集成开发环境(IDE),为我们提供了所有机器学习功能,使我们能够在一个窗口中完成。在运行SageMaker JumpStart之前,我们需要设置SageMaker Studio。如果您已经有自己的版本的SageMaker Studio在运行,您可以跳过这一步。
在我们使用任何AWS服务之前,首先要确保我们已经注册并创建了AWS账户。然后我们创建一个管理员用户和一个组。有关这两个步骤的详细说明,请参阅设置Amazon SageMaker先决条件。
下一步是创建一个SageMaker域。域设置所有的存储并允许您添加用户以访问SageMaker。有关更多信息,请参阅加入Amazon SageMaker域。此演示是在AWS区域us-east-1中创建的。
最后,您启动SageMaker Studio。对于本文,我们建议启动一个用户配置文件应用程序。有关说明,请参阅启动Amazon SageMaker Studio。
要运行此SageMaker JumpStart解决方案并将基础架构部署到您的AWS账户中,您需要创建一个活动的SageMaker Studio实例(请参阅加入Amazon SageMaker Studio)。当您的实例准备好后,使用SageMaker JumpStart中的说明来启动解决方案。解决方案的工件包含在此GitHub存储库中供参考。
启动SageMaker Jumpstart解决方案
要开始使用解决方案,请完成以下步骤:
- 在SageMaker Studio控制台上,选择JumpStart。
- 在解决方案选项卡上,选择车队的预测性维护。
- 选择启动。
部署解决方案需要几分钟时间。
- 解决方案部署完成后,选择打开笔记本。
如果提示选择内核,请为此解决方案中的所有笔记本选择PyTorch 1.8 Python 3.6。
解决方案预览
我们首先在0_demo.ipynb
笔记本上进行工作。在这个笔记本中,您可以快速预览完成此解决方案的完整笔记本后的结果。
选择运行和运行所有单元格来在SageMaker Studio中运行所有单元格(或在SageMaker笔记本实例中选择单元格和运行所有)。您可以依次运行每个笔记本中的所有单元格。确保所有单元格完成处理后再转到下一个笔记本。
此解决方案依赖于配置文件来运行所提供的AWS资源。我们按以下方式生成该文件:
import boto3
import os
import json
client = boto3.client('servicecatalog')
cwd = os.getcwd().split('/')
i= cwd.index('S3Downloads')
pp_name = cwd[i + 1]
pp = client.describe_provisioned_product(Name=pp_name)
record_id = pp['ProvisionedProductDetail']['LastSuccessfulProvisioningRecordId']
record = client.describe_record(Id=record_id)
keys = [ x['OutputKey'] for x in record['RecordOutputs'] if 'OutputKey' and 'OutputValue' in x]
values = [ x['OutputValue'] for x in record['RecordOutputs'] if 'OutputKey' and 'OutputValue' in x]
stack_output = dict(zip(keys, values))
with open(f'/root/S3Downloads/{pp_name}/stack_outputs.json', 'w') as f:
json.dump(stack_output, f)
我们有一些样本时间序列输入数据,包括车辆的电池电压和电池电流随时间变化的数据。接下来,我们加载和可视化样本数据。如下所示的屏幕截图显示,电压和电流值在Y轴上,读数(记录了19个读数)在X轴上。
我们之前在这些电压和电流数据上训练了一个模型,用于预测车辆故障的概率,并将该模型部署为SageMaker中的端点。我们将使用一些样本数据调用此端点,以确定下一个时间段内的故障概率。
根据样本输入数据,预测的故障概率为45.73%。
要进入下一阶段,请选择点击此处继续。
介绍和解决方案概述
1_introduction.ipynb
笔记本提供了解决方案和阶段的概述,以及配置文件的内容定义、数据采样周期、训练和测试样本数量、参数、位置和生成内容的列名的介绍。
在您查看完这个笔记本后,您可以进入下一阶段。
准备样本数据集
我们在2_data_preparation.ipynb
笔记本中准备了一个样本数据集。
我们首先为这个解决方案生成配置文件:
import boto3
import os
import json
client = boto3.client('servicecatalog')
cwd = os.getcwd().split('/')
i= cwd.index('S3Downloads')
pp_name = cwd[i + 1]
pp = client.describe_provisioned_product(Name=pp_name)
record_id = pp['ProvisionedProductDetail']['LastSuccessfulProvisioningRecordId']
record = client.describe_record(Id=record_id)
keys = [ x['OutputKey'] for x in record['RecordOutputs'] if 'OutputKey' and 'OutputValue' in x]
values = [ x['OutputValue'] for x in record['RecordOutputs'] if 'OutputKey' and 'OutputValue' in x]
stack_output = dict(zip(keys, values))
with open(f'/root/S3Downloads/{pp_name}/stack_outputs.json', 'w') as f:
json.dump(stack_output, f)
import os
from source.config import Config
from source.preprocessing import pivot_data, sample_dataset
from source.dataset import DatasetGenerator
config = Config(filename="config/config.yaml", fetch_sensor_headers=False)
config
配置属性如下:
fleet_info_fn=data/example_fleet_info.csv
fleet_sensor_logs_fn=data/example_fleet_sensor_logs.csv
vehicle_id_column=vehicle_id
timestamp_column=timestamp
target_column=target
period_ms=30000
dataset_size=25000
window_length=20
chunksize=10000
processing_chunksize=2500
fleet_dataset_fn=data/processed/fleet_dataset.csv
train_dataset_fn=data/processed/train_dataset.csv
test_dataset_fn=data/processed/test_dataset.csv
period_column=period_ms
您可以定义自己的数据集,或使用我们的脚本生成一个样本数据集:
if should_generate_data:
fleet_statistics_fn = "data/generation/fleet_statistics.csv"
generator = DatasetGenerator(fleet_statistics_fn=fleet_statistics_fn,
fleet_info_fn=config.fleet_info_fn,
fleet_sensor_logs_fn=config.fleet_sensor_logs_fn,
period_ms=config.period_ms,
)
generator.generate_dataset()
assert os.path.exists(config.fleet_info_fn), "Please copy your data to {}".format(config.fleet_info_fn)
assert os.path.exists(config.fleet_sensor_logs_fn), "Please copy your data to {}".format(config.fleet_sensor_logs_fn)
您可以将传感器数据和车队车辆数据合并在一起:
pivot_data(config)
sample_dataset(config)
现在我们可以进行数据可视化。
可视化我们的示例数据集
我们在3_data_vizualization.ipynb
中可视化我们的示例数据集。此解决方案依赖于配置文件来运行配置的AWS资源。让我们生成与之前的笔记本类似的文件。
以下屏幕截图显示了我们的数据集。
接下来,让我们构建数据集:
train_ds = PMDataset_torch(
config.train_dataset_fn,
sensor_headers=config.sensor_headers,
target_column=config.target_column,
standardize=True)
properties = train_ds.vehicle_properties_headers.copy()
properties.remove('vehicle_id')
properties.remove('timestamp')
properties.remove('period_ms')
现在数据集已经准备好了,让我们来可视化数据统计信息。以下屏幕截图显示了基于车辆制造商、发动机类型、车辆类别和型号的数据分布。
比较日志数据,让我们看一个关于不同年份的Make E和C(随机)的平均电压的示例。
电压和电流的平均值在Y轴上,读数的数量在X轴上。
- log_target的可能值:[‘make’, ‘model’, ‘year’, ‘vehicle_class’, ‘engine_type’]
- log_target的随机分配值:
make
- log_target的随机分配值:
- log_target_value1的可能值:[‘Make A’, ‘Make B’, ‘Make E’, ‘Make C’, ‘Make D’]
- log_target_value1的随机分配值:
Make B
- log_target_value1的随机分配值:
- log_target_value2的可能值:[‘Make A’, ‘Make B’, ‘Make E’, ‘Make C’, ‘Make D’]
- log_target_value2的随机分配值:
Make D
- log_target_value2的随机分配值:
根据上述条件,我们假设log_target: make
,log_target_value1: Make B
和log_target_value2: Make D
以下图表展示了日志数据的平均值。
以下图表显示了不同传感器日志值与电压和电流之间的示例。
在我们的示例数据集上训练模型以检测故障
在4_model_training.ipynb
笔记本中,我们在示例数据集上训练模型以检测故障。
让我们生成与之前的笔记本类似的配置文件,然后进行训练配置:
sage_session = sagemaker.session.Session()
s3_bucket = sagemaker_configs["S3Bucket"]
s3_output_path = 's3://{}/'.format(s3_bucket)
print("S3 bucket path: {}".format(s3_output_path))
# 在本地模式下运行此计算机或作为SageMaker TrainingJob运行
local_mode = False
if local_mode:
instance_type = 'local'
else:
instance_type = sagemaker_configs["SageMakerTrainingInstanceType"]
role = sagemaker.get_execution_role()
print("Using IAM role arn: {}".format(role))
# 仅从SageMaker笔记本实例运行
if local_mode:
!/bin/bash ./setup.sh
cpu_or_gpu = 'gpu' if instance_type.startswith('ml.p') else 'cpu'
我们现在可以定义数据并启动超参数优化:
%%time
estimator = PyTorch(entry_point="train.py",
source_dir='source',
role=role,
dependencies=["source/dl_utils"],
instance_type=instance_type,
instance_count=1,
output_path=s3_output_path,
framework_version="1.5.0",
py_version='py3',
base_job_name=job_name_prefix,
metric_definitions=metric_definitions,
hyperparameters= {
'epoch': 100, # 根据您的需求进行调整
'target_column': config.target_column,
'sensor_headers': json.dumps(config.sensor_headers),
'train_input_filename': os.path.basename(config.train_dataset_fn),
'test_input_filename': os.path.basename(config.test_dataset_fn),
}
)
if local_mode:
estimator.fit({'train': training_data, 'test': testing_data})
%%time
tuner = HyperparameterTuner(estimator,
objective_metric_name='test_auc',
objective_type='Maximize',
hyperparameter_ranges=hyperparameter_ranges,
metric_definitions=metric_definitions,
max_jobs=max_jobs,
max_parallel_jobs=max_parallel_jobs,
base_tuning_job_name=job_name_prefix)
tuner.fit({'train': training_data, 'test': testing_data})
分析我们训练的模型的结果
在 5_results_analysis.ipynb
笔记本中,我们从超参数调优作业中获取数据,可视化所有作业的指标以确定最佳作业,并为最佳训练作业构建一个终端节点。
让我们生成类似于之前笔记本中的配置文件,并可视化所有作业的指标。下面的图表显示测试准确率与 epoch 的关系。
下面的截图显示我们运行的超参数调优作业。
您现在可以根据测试准确率(红色)可视化来自四个训练作业中最佳训练作业的数据。
正如下面的截图所示,随着 epoch 的增加,测试损失减小,AUC 和准确率增加。
根据可视化结果,我们现在可以为最佳训练作业构建一个终端节点:
%%time
role = sagemaker.get_execution_role()
model = PyTorchModel(model_data=model_artifact,
role=role,
entry_point="inference.py",
source_dir="source/dl_utils",
framework_version='1.5.0',
py_version = 'py3',
name=sagemaker_configs["SageMakerModelName"],
code_location="s3://{}/endpoint".format(s3_bucket)
)
endpoint_instance_type = sagemaker_configs["SageMakerInferenceInstanceType"]
predictor = model.deploy(initial_instance_count=1, instance_type=endpoint_instance_type, endpoint_name=sagemaker_configs["SageMakerEndpointName"])
def custom_np_serializer(data):
return json.dumps(data.tolist())
def custom_np_deserializer(np_bytes, content_type='application/x-npy'):
out = np.array(json.loads(np_bytes.read()))
return out
predictor.serializer = custom_np_serializer
predictor.deserializer = custom_np_deserializer
构建终端节点后,我们可以通过传递示例传感器日志来测试预测器:
import botocore
config = botocore.config.Config(read_timeout=200)
runtime = boto3.client('runtime.sagemaker', config=config)
data = np.ones(shape=(1, 20, 2)).tolist()
payload = json.dumps(data)
response = runtime.invoke_endpoint(EndpointName=sagemaker_configs["SageMakerEndpointName"],
ContentType='application/json',
Body=payload)
out = json.loads(response['Body'].read().decode())[0]
print("给定示例输入数据,故障发生的预测概率为 {:0.2f}%".format(100*(1.0-out[0])))
给定样本输入数据,故障的预测概率为34.60%。
清理
完成此解决方案后,请确保删除所有不需要的AWS资源。在车辆车队的预测性维护页面上,在删除解决方案下,选择删除所有资源以删除与解决方案相关联的所有资源。
您需要手动删除在此笔记本中可能创建的任何额外资源。一些示例包括额外的S3存储桶(到解决方案的默认存储桶)和额外的SageMaker端点(使用自定义名称)。
自定义解决方案
我们的解决方案很容易进行自定义。要修改输入数据可视化,请参考sagemaker/3_data_visualization.ipynb。要自定义机器学习,请参考sagemaker/source/train.py和sagemaker/source/dl_utils/network.py。要自定义数据集处理,请参考sagemaker/1_introduction.ipynb以了解如何定义配置文件。
此外,您可以更改配置文件中的配置。默认配置如下:
fleet_info_fn=data/example_fleet_info.csv
fleet_sensor_logs_fn=data/example_fleet_sensor_logs.csv
vehicle_id_column=vehicle_id
timestamp_column=timestamp
target_column=target
period_ms=30000
dataset_size=10000
window_length=20
chunksize=10000
processing_chunksize=1000
fleet_dataset_fn=data/processed/fleet_dataset.csv
train_dataset_fn=data/processed/train_dataset.csv
test_dataset_fn=data/processed/test_dataset.csv
period_column=period_ms
配置文件具有以下参数:
fleet_info_fn
,fleet_sensor_logs_fn
,fleet_dataset_fn
,train_dataset_fn
和test_dataset_fn
定义了数据集文件的位置vehicle_id_column
,timestamp_column
,target_column
和period_column
定义了列的标题dataset_size
,chunksize
,processing_chunksize
,period_ms
和window_length
定义了数据集的属性
结论
在本文中,我们向您展示了如何使用SageMaker JumpStart训练和部署模型来预测车辆车队故障概率。该解决方案基于ML和深度学习模型,允许使用各种输入数据,包括任何时变传感器数据。因为每辆车上都有不同的遥测数据,您可以根据您拥有的数据的频率和类型对提供的模型进行微调。
要了解有关使用SageMaker JumpStart可以做什么的更多信息,请参考以下内容:
- 使用Amazon SageMaker JumpStart进行视觉检查自动化
- 使用Amazon SageMaker JumpStart运行自动模型调整
- 使用Amazon SageMaker JumpStart在AWS上开始生成AI
资源
- Amazon SageMaker开发者指南
- SageMaker JumpStart开发者指南
- 使用SageMaker进行自动模型调整
- SageMaker JumpStart预测性维护解决方案