GS Caltex 如何用 Amazon SageMaker Canvas 与 Pipelines 实现生产级 MLOps（第一部分）

GS Caltex 使用 Amazon SageMaker Canvas 和 Amazon SageMaker Pipelines 将工艺工程师的模型连接到生产工作流。这家韩国能源和化工公司将模型开发与运维工作分配给技能互补的工程师。

工艺工程师无需编写代码，即可在 SageMaker Canvas 中利用其工艺知识构建并验证模型。机器学习（ML）工程师则配置 SageMaker Pipelines 工作流，用于部署和定时推理。

将资产从 SageMaker Canvas 迁移到 SageMaker Pipelines 仍需进行集成工作。ML 工程师必须为生产环境设置数据路径、提供模型配置，并连接预处理和推理作业。GS Caltex 在复用 SageMaker Canvas 中所创建资产的同时，实现了这一交接的自动化。

GS Caltex 表示，其 ML 工程师现在可在数小时内完成模型部署，而此前大约需要一周。该公司每 30 分钟运行一次批量推理，从而减少了对持续运行端点的需求。GS Caltex 表示，这已显著降低了其推理成本。

本文介绍 GS Caltex 如何将 SageMaker Canvas 连接到 SageMaker Pipelines，并在生产环境中复用经过验证的预处理流程和模型。

解决方案概述

机器学习运维（MLOps）工作流包含两条路径：为新模型版本做好部署准备，以及使用已部署模型运行批量推理。

从经过验证的模型到可重复的工作流

您可以使用 SageMaker Canvas 以可视化方式准备数据、构建模型并生成批量预测。您还可以从 SageMaker Canvas 将经过验证的模型注册到 Amazon SageMaker Model Registry。GS Caltex 构建了一个单独的交接流程，将这些资产转化为可执行的 SageMaker Pipelines 步骤。

工艺工程师了解工艺背后的变量和数据，并评估模型的预测是否有用。ML 工程师则将生成的资产转化为生产作业，并管理模型版本和执行历史。

参考架构

下图展示了工艺工程师如何将 SageMaker Canvas 资产交接给 ML 工程师，用于部署和定时推理。ML 工程师使用 Amazon Simple Storage Service（Amazon S3）存储预处理流程，使用 Model Registry 管理模型版本，然后通过 Amazon EventBridge 调度管道运行。

图 1：SageMaker Canvas 资产连接到 SageMaker Pipelines 工作流，用于定时批量推理

ML 工程师从 SageMaker Canvas 接收预处理流程和模型，并审查数据格式和预测结果。审查完成后，他们在 Model Registry 中批准该模型并进行部署。这一准备工作在部署新模型版本时进行。后续推理运行会复用已部署的模型。

Amazon EventBridge 启动每次计划运行。SageMaker Pipelines 从 Amazon Simple Storage Service (Amazon S3) 准备输入数据，运行预处理和 SageMaker Batch Transform，并将结果存储在 S3 中。ML 工程师可以检查步骤状态和 Amazon CloudWatch 日志来监控工作流。

生产管道复用 SageMaker Canvas 中创建的预处理逻辑和已训练模型。ML 工程师配置输入和输出路径，设置执行顺序，并编写将审批和调度连接到工作流的代码。

先决条件

在开始之前，请准备以下内容：

• 在 SageMaker Canvas 中构建并评估的模型，以及其 Data Wrangler 预处理流程
• 用于推理的 CSV 数据以及用于存储文件的 S3 存储桶。
• 一个 SageMaker 执行角色和一个具有启动管道权限的 Amazon EventBridge 角色
• 安装了 Boto3 和 SageMaker Python SDK 2.x 的 Python 环境。

完整的参考实现可在 GitHub 仓库中获取。本文中的代码示例侧重于组件之间的连接。该仓库使用示例贷款分类模型，因此请调整配置以匹配您模型的输入和输出格式。

技术实现

以下步骤涵盖预处理复用、模型审批和部署、管道构建以及计划推理。

1. 使用 SageMaker Processing 运行 SageMaker Canvas 预处理流程

在 SageMaker Canvas 中使用 Data Wrangler 创建的预处理流程会记录数据源和转换步骤。要设置参考实现，请使用 SageMaker Processing 将数据导出到 Amazon S3，然后检索作业设置以及该作业使用的 .flow 文件。

在导出设置中，关闭 Auto job configuration 并选择 SageMaker Processing。导出开始后，在 Amazon SageMaker AI 控制台上打开 Processing 作业详细信息。记录容器映像 URI、名为 flow 的输入的 S3 URI、执行角色以及输出设置。使用这些值来配置运行 .flow 文件的管道步骤。

SageMaker 执行角色必须具有读取所需 S3 数据、写入结果以及访问容器映像的权限。根据需要处理的数据量选择实例类型和数量。

保留由流程工程师验证的预处理逻辑，并针对生产环境调整数据源、路径和其他设置。对照原始作业配置检查容器映像和输出节点。

部分 .flow 文件包含执行元数据，例如 internal_metadata.dw_job.container_uri。请检查实际的 Processing 作业设置，而不是假设每个导出都包含此字段。如果将工作流迁移到另一个 AWS 区域，请确认容器镜像在该区域可用。

参考实现按 runs/<execution-id>/ 下的执行来区分中间输出和最终输出。其 PrepareInput 步骤将数据写入由 .flow 文件引用的固定 S3 位置。防止共享该输入路径的执行发生重叠。

若要为每次执行隔离输入数据，可添加一个准备步骤，在 .flow 文件的副本中更新源 URI。声明 SageMaker Pipelines 参数并不会自动替换文件内的 URI。

2. 处理训练输入与推理输入之间的差异

目标列是训练数据与推理数据之间的一项差异。使用训练数据构建的流程可能包含期望输入架构中存在目标列的转换。推理数据不包含该已知结果。

参考实现在预处理之前添加了一个 PrepareInput 步骤。如果缺少目标列，该步骤会用一个占位值添加它。预处理之后，DropTargetColumn 会移除该列。使占位值的数据类型和列位置与训练架构匹配，并仅按原始顺序将预期特征传递给模型。

PrepareInput -> PreprocessNewData -> DropTargetColumn
             -> BatchPredict -> AssembleResults

仅当占位目标不影响特征值时，才使用此方法。如果流程按目标筛选行、从目标派生特征或应用目标编码，则需要额外处理。在这些情况下，请分离推理所需的转换和已学习状态。

验证应覆盖列顺序和数量、数据类型、缺失值处理以及行数。移除目标后，检查剩余列是否与训练期间使用的特征顺序匹配。

3. 注册、批准并部署模型

在 Model Registry 中注册已验证的 SageMaker Canvas 模型版本。注册的版本可以包含模型文件、推理配置和评估结果。ML 工程师在更改批准状态之前会审查此信息。

模型文件（例如 model.tar.gz）必须与正确的推理容器和环境设置一起使用。Model Registry 中的模型包会将这些信息组合在一起。在此批处理架构中，部署会从已批准的包创建 SageMaker 模型资源，供 Batch Transform 引用。

以下示例检查一个已批准的包，然后创建模型资源。部署期间提供 model_package_arn、model_name 和 sagemaker_role_arn。此示例为每个版本使用唯一的模型名称。

import boto3

sm_client = boto3.client("sagemaker")
package = sm_client.describe_model_package(
    ModelPackageName=model_package_arn
)

if package["ModelPackageStatus"] != "Completed":
    raise RuntimeError("Model package is not ready")
if package["ModelApprovalStatus"] != "Approved":
    raise RuntimeError("Model package is not approved")

sm_client.create_model(
    ModelName=model_name,
    ExecutionRoleArn=sagemaker_role_arn,
    PrimaryContainer={"ModelPackageName": model_package_arn},
)

Model Registry 会记录批准状态，但不会对每个推理请求强制执行批准检查。请在部署代码中检查该状态，并在管道配置中保留所选包的 ARN 和模型名称。如果批准被撤销，请更新使用该模型的计划和执行配置。

注册和部署在模型版本发布时进行。每 30 分钟运行一次的管道使用该模型创建 Batch Transform 作业，因此无需在每次运行时重新注册模型。

4. 连接预处理、推理和结果组装

推理管道包含五个步骤。

在连接各步骤之前配置每个作业。在以下示例中，prepare_args、preprocess_args、drop_target_args 和 assemble_args 保存由使用 PipelineSession 的处理器上的 run() 调用返回的步骤参数。这些处理器和 transformer 共享 pipeline_session 对象。使用先前部署的 model_name 配置 transformer。

from sagemaker.workflow.pipeline import Pipeline
from sagemaker.workflow.steps import ProcessingStep, TransformStep

prepare = ProcessingStep(
    name="PrepareInput", step_args=prepare_args
)
preprocess = ProcessingStep(
    name="PreprocessNewData",
    step_args=preprocess_args,
    depends_on=[prepare],
)
drop_target = ProcessingStep(
    name="DropTargetColumn",
    step_args=drop_target_args,
    depends_on=[preprocess],
)

# The DropTargetColumn job defines an output named "features".
features_uri = (
    drop_target.properties.ProcessingOutputConfig
    .Outputs["features"].S3Output.S3Uri
)
predict = TransformStep(
    name="BatchPredict",
    step_args=transformer.transform(
        data=features_uri,
        content_type="text/csv",
        split_type="Line",
    ),
)
assemble = ProcessingStep(
    name="AssembleResults",
    step_args=assemble_args,
    depends_on=[predict],
)

pipeline = Pipeline(
    name="canvas-process-inference",
    steps=[prepare, preprocess, drop_target, predict, assemble],
    sagemaker_session=pipeline_session,
)
pipeline_arn = pipeline.upsert(role_arn=sagemaker_role_arn)["PipelineArn"]

以下执行图展示了示例贷款模型推理工作流中的五个步骤。它们准备输入、应用预处理、移除目标列、运行批量推理并汇总结果。截取此屏幕截图时，执行处于 Executing 状态。

图 2：示例贷款模型的 SageMaker Pipelines 执行图

使用 depends_on 控制执行顺序，并通过步骤输出属性传递数据。例如，assemble_args 必须接收预处理数据的位置和 predict.properties.TransformOutput.S3OutputPath。输出名称 "features" 属于 DropTargetColumn。Data Wrangler 步骤使用其预处理流中定义的输出 ID。

在参考实现中，AssembleResults 按行位置合并预处理数据和预测结果。在生产环境中使用之前，请添加检查以确保文件顺序和行数匹配，如果不匹配则停止生成结果。如果预处理会排序或删除行，请维护源记录 ID 与预测结果之间的显式映射。

5. 每 30 分钟调度一次推理

GS Caltex 每 30 分钟运行一次推理。以下示例使用 Amazon EventBridge 规则来调度 SageMaker Pipelines 执行。在附加调度之前，请创建管道并通过手动运行进行验证。

import boto3

events_client = boto3.client("events")
rule_name = "canvas-process-inference-every-30-minutes"

events_client.put_rule(
    Name=rule_name,
    ScheduleExpression="rate(30 minutes)",
    State="ENABLED",
)

result = events_client.put_targets(
    Rule=rule_name,
    Targets=[{
        "Id": "canvas-process-inference",
        "Arn": pipeline_arn,
        "RoleArn": eventbridge_role_arn,
    }],
)
if result["FailedEntryCount"]:
    raise RuntimeError(result["FailedEntries"])

将 eventbridge_role_arn 设置为信任 Amazon EventBridge 服务主体 events.amazonaws.com 且对目标管道具有 sagemaker:StartPipelineExecution 权限的角色。pipeline_arn 值标识上一步中创建的管道。

表达式 rate(30 minutes) 每 30 分钟调度一次运行。要在整点和半点运行，请使用 cron(0/30 * * * ? *)。在输入准备步骤中确定每次运行处理的数据窗口，并将其与执行一起记录。

运维注意事项

在启用计划运行之前，请检查总成本、执行时间以及对无效或不匹配数据的处理。

评估整个工作流的批量成本

Amazon SageMaker Pipelines 是一种无服务器工作流编排服务。它管理执行顺序和依赖关系，无需你预置或维护工作流服务器。处理和批量转换作业使用单独的计算资源，并根据实例类型和持续时间计费。

预置的实时端点非常适合需要始终低延迟响应的请求。批量转换适合可以一起处理一组记录的工作负载。作业完成后，它会释放作业的计算资源，因此你可以运行预测，而无需为每个模型持续保持推理端点运行。

30 分钟的计划每天会产生 48 次计划运行。估算总运行时间时，需计入预处理、实例启动、推理和后处理。成本比较还应考虑工作流使用的 Processing、S3、CloudWatch 及其他服务。

使用实测用量评估每个工作负载的成本，包括配套服务以及每个计划作业的完整运行时间。

检查运行时间与重叠执行

每 30 分钟启动一次运行的计划，并不能保证每次运行都在该间隔内完成。请使用有代表性的数据测量完整流水线运行时间，包括实例启动以及重试失败作业所需的时间。

确定如何处理在上一次执行尚未完成时到来的计划运行。控制流水线内的并行步骤并不能防止不同的流水线执行发生重叠。请检查活动执行、防止对同一输入窗口重复处理，并为每次运行分离输出路径。

记录每次运行的输入数据窗口、预处理流程版本、模型包 ARN、流水线执行 ARN 和输出位置。这些详细信息有助于复现结果和排查错误。

验证输出与失败行为

在启用计划之前，请完成以下检查：

1. 使用验证数据集运行 SageMaker Canvas 和 Processing。比较输出列、数据类型、行数以及关键转换值。
2. 检查添加占位目标是否不会改变特征值，以及该目标是否已从模型输入中移除。
3. 使用已批准的模型版本运行 Batch Transform，并验证源记录与预测之间的关联。
4. 确认无效的输入架构或预处理失败会阻止下游推理和结果发布。
5. 测量运行时间和成本，测试重叠执行以及同一输入窗口的重复处理，然后启用 30 分钟的计划。

为作业失败和处理延迟添加 CloudWatch 告警，并根据需要添加数据质量或模型监控。根据实测运行时间和数据特征设置初始阈值，并在模型或数据量发生变化时重新审视这些阈值。

采用方法与成果

通过分阶段推广，团队可以在将交接流程标准化到各个模型之前先对其进行验证。以下各节介绍了一种采用方法以及 GS Caltex 报告的成果。

从模型开发转向常规运营

首先为一个模型验证通往生产的路径，然后标准化在各次部署中重复出现的步骤。以下序列是基于 GS Caltex 经验提出的建议采用方法。

随着自动化范围的扩大，工艺工程师会持续评估这些预测对工艺是否有用。ML 工程师负责管理管道的执行和变更。

GS Caltex 的成果

GS Caltex 减少了将 SageMaker Canvas 模型接入生产所需的重复工作。ML 工程师现在能在几小时内完成模型部署，而此前大约需要一周。他们还能复用工艺工程师在 SageMaker Canvas 中验证过的资产。

工艺工程师看到自己的模型投入生产后，开始在其他工艺中提出应用方案。ML 工程师用于重复集成工作的时间减少，可以专注于管道的运行和监控。

清理测试资源

对于测试部署，先禁用 Amazon EventBridge 规则并移除其目标。在活动中的管道执行和作业完成后，删除测试管道及其 SageMaker 模型资源。

如果你为测试创建了实时端点，也要将其删除。从 S3 中移除输入、中间和预测文件，并根据你的保留策略清理 CloudWatch 日志。在删除模型文件和 Model Registry 版本之前，检查是否有其他执行或审计要求仍依赖它们。

结论

在 GS Caltex，工艺工程师在 SageMaker Canvas 中创建并验证模型，ML 工程师使用这些资产在 SageMaker Pipelines 中运行生产工作流。这种交接减少了重复的集成工作。GS Caltex 报告称，ML 工程师现在能在几小时内完成模型部署，而此前大约需要一周。

交接内容除了模型文件外，还需要包括预处理流程、输入架构、推理配置和审批流程。验证工作流能复现预期结果，并能按所需计划可靠运行。

要尝试这种方法，请从你在 SageMaker Canvas 中验证过的一个模型开始，并使用 GitHub 仓库。有关定义和运行工作流的详细信息，请参阅 SageMaker Pipelines 开发者指南。通过手动执行检查预处理和批量预测结果，然后按你的应用所需的间隔安排运行。

在第 2 部分中，我们将展示 GS Caltex 如何使用 Amazon Bedrock 自动将现有模型迁移到该管道上。

关于作者

WonGu Rho

WonGu 是 GS Caltex DAX Lab 的数据分析师。他专注于数据分析、机器学习模型开发与运维、模型性能优化和 MLOps。此外，他还通过设计和集成企业 AI 智能体、工作流和数据管道，推动业务团队的运营效率和数据驱动决策。

Hyunsoo Kim, Ph.D.

Hyunsoo 是 Amazon Web Services 的 APJ 高级 GTM 生成式 AI 专家解决方案架构师。他常驻首尔，帮助亚太地区的客户设计生成式 AI 解决方案，包括 AI 智能体、高级 RAG、模型优化和 MLOps。

Sang Hyun Kim

Sang 是 AWS Korea 的高级客户主管。他帮助能源和制造业的客户采用云技术并将 AI 付诸实践。他与 AWS 专家和合作伙伴合作，支持客户从概念验证到生产部署以及更广泛的应用。

Joonhyung Kim

Joonhyung 是 AWS Korea 的高级解决方案架构师，也是物理 AI 和空间计算专家。他帮助客户在 AWS 上构建和运营数字孪生、3D 以及 AR/VR/XR 应用，重点关注交付速度和成本效率。他的经验涵盖生成式 AI、机器学习、数据和云架构。