Article · AI translation
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 工程师现在可以在几小时内完成模型部署，而此前大约需要 1 周。该公司每 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 中准备的内容 在生产工作流中管理的内容 数据准备 使用 Data Wrangler 构建的预处理流程 Amazon SageMaker Processing 中的预处理作业 模型开发 训练好的模型和评估结果 Model Registry 中的模型版本和审批状态 预测 对模型预测结果的验证 使用获批模型每 30 分钟进行一次批量推理 运营 单个模型的开发和验证 步骤执行历史、存储的结果和故障处理 参考架构 下图展示了工艺工程师如何将 SageMaker Canvas 资产移交给 ML 工程师，以便部署和进行定时推理。ML 工程师使用 Amazon Simple Storage Service（Amazon S3）存放预处理流程，使用 Model Registry 管理模型版本，然后通过 Amazon EventBridge 安排流水线定时运行。图 1：连接到 SageMaker Pipelines 工作流、用于定时批量推理的 SageMaker Canvas 资产 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 文件。在导出设置中，关闭“自动作业配置”并选择 SageMaker Processing。导出开始后，在 Amazon SageMaker AI 控制台中打开 Processing 作业详细信息。记录容器镜像 URI、名为 flow 的输入的 S3 URI、执行角色以及输出设置。使用这些值配置运行 .flow 文件的流水线步骤。SageMaker 执行角色必须有权读取所需的 S3 数据、写入结果并访问容器镜像。根据需要处理的数据量选择实例类型和数量。保留由流程工程师验证过的预处理逻辑，并根据生产环境调整数据源、路径和其他设置。对照原始作业配置检查容器镜像和输出节点。检查项 操作 输入位置 检查 .flow 文件中的数据源所引用的 S3 URI。将原始数据上传到其他位置不会更新这一引用。容器镜像 使用在导出到 S3 期间创建的 Processing 作业所用的镜像，并固定到能够复现已验证转换的版本。输出 ID 使用导出配置中的输出节点和输出名称。如果所需格式为 {node_id}.{output_name}，请严格匹配。转换中学到的值 保留训练期间学到的值，例如缩放和编码参数。不要针对每个推理批次重新拟合这些参数。验证 对于相同的输入，比较 SageMaker Canvas 和 Processing 生成的列、数据类型和转换后的值。某些 .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. 注册、批准和部署模型 将经过验证的 SageMaker Canvas 模型版本注册到 Model Registry。注册的版本可以包含模型文件、推理配置和评估结果。机器学习工程师会在更改批准状态前审查这些信息。模型文件（例如 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. 连接预处理、推理和结果组装 推理管道有五个步骤。步骤 功能 输入和输出 PrepareInput 如有需要，添加占位目标列 新推理数据 -> 已添加目标列的数据 PreprocessNewData 运行 SageMaker Canvas 预处理流程 准备好的输入和 .flow -> 转换后的数据 DropTargetColumn 移除目标列和 CSV 表头 转换后的数据 -> 无表头的特征 CSV BatchPredict 使用已批准的模型运行批量推理 特征 CSV -> 预测文件 AssembleResults 将预处理后的数据与预测结果合并 预处理后的数据和预测文件 -> 最终结果 CSV 在连接各步骤之前配置每个作业。在以下示例中，prepare_args、preprocess_args、drop_target_args 和 assemble_args 保存使用 PipelineSession 的处理器调用 run() 后返回的步骤参数。这些处理器和转换器共用 pipeline_session 对象。使用先前部署的 model_name 配置转换器。 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], ) # DropTargetColumn 作业定义了一个名为 "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"] 下方的执行图展示了示例贷款模型推理工作流的 5 个步骤：准备输入、应用预处理、移除目标列、运行批量推理，以及汇总结果。截取此屏幕截图时，执行状态为 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 是一种无服务器工作流编排服务。它管理执行顺序和依赖关系，无需您预置或维护工作流服务器。Processing 和 Batch Transform 作业使用独立的计算资源，并根据实例类型和使用时长计费。对于需要持续低延迟响应的请求，预置的实时端点是合适的选择。Batch Transform 适用于能够将一组记录一起处理的工作负载。 作业完成后，它会释放该作业的计算资源，因此无需为每个模型持续运行推理端点，也能执行预测。考量因素：预置的实时端点；使用 Batch Transform 的 SageMaker Pipelines。资源生命周期：实例持续可用以处理请求；每项作业分别预置并释放资源。成本驱动因素：端点运行时间和容量；作业时长、频率、实例数量及配套服务。适用工作负载：需要即时响应和持续处理在线请求；按计划间隔处理累积的数据。添加模型：增加端点容量和持续托管成本；增加处理工作、作业时长及并发限制。每 30 分钟执行一次的计划每天会产生 48 次计划运行。估算总运行时间时，应计入预处理、实例启动、推理和后处理。成本比较还应计入工作流程使用的 Processing、S3、CloudWatch 和其他服务。根据各项工作负载的实测用量评估其成本，包括配套服务和每次计划作业的完整运行时间。检查运行时间和重叠执行。每 30 分钟启动一次运行，并不保证每次运行都能在这段时间内完成。使用具有代表性的数据测量整个流水线的运行时间，包括实例启动时间和重试失败作业所需的时间。决定如何处理前一次执行尚未结束时到来的计划运行。控制流水线内部步骤的并行执行，并不能防止不同流水线执行之间发生重叠。检查正在进行的执行，防止对同一输入时间窗口进行重复处理，并为每次运行设置独立的输出路径。记录每次运行的输入数据时间窗口、预处理流程版本、模型包 ARN、流水线执行 ARN 和输出位置。这些信息有助于重现结果和调查错误。验证输出和故障处理行为。启用计划之前，完成以下检查：让验证数据集经过 SageMaker Canvas 和 Processing。比较输出列、数据类型、行数和关键转换值。检查添加占位目标值是否会改变特征值，并确认已从模型输入中移除目标值。使用获批的模型版本运行 Batch Transform，并验证源记录与预测结果之间的对应关系。确认无效的输入数据架构或预处理失败会阻止后续推理和结果发布。测量运行时间和成本，测试重叠执行以及对同一输入时间窗口的重新处理，然后启用每 30 分钟运行一次的计划。针对作业失败和处理延迟添加 CloudWatch 告警，并根据需要添加数据质量或模型监控。根据实测运行时间和数据特征设置初始阈值，并在模型或数据量发生变化时重新评估。采用方式与成果。通过分阶段推出，团队可以先验证交接流程，再将其标准化并应用于多个模型。以下各节介绍一种采用方式以及 GS Caltex 报告的成果。从模型开发转向常规运营。首先验证一个模型的上线流程，然后将各次部署中重复的步骤标准化。以下步骤是根据 GS Caltex 的经验提出的一种建议采用方式。 阶段 目标 范围 Crawl 在生产环境中复现开发结果 验证 SageMaker Canvas 模型，将其注册到 Model Registry，并手动运行流水线 Walk 建立可靠的定时运行机制 检查审批情况，按要求的时间间隔安排运行，添加故障警报，并按执行实例管理输出 Run 在不同模型间复用运行流程 引入通用模板、持续集成和持续交付（CI/CD）、变更验证，以及移交运行职责的程序 随着自动化范围扩大，工艺工程师继续评估预测结果是否对工艺有用。机器学习工程师管理流水线的执行和变更。GS Caltex 的成果 GS Caltex 减少了将 SageMaker Canvas 模型接入生产环境所需的重复工作。机器学习工程师现在可在几小时内完成模型部署，而此前约需一周。他们还复用工艺工程师在 SageMaker Canvas 中验证过的资产。工艺工程师看到自己的模型投入生产后，开始提议将其应用于其他工艺。机器学习工程师减少了重复集成工作所花的时间，得以专注于流水线的运行和监控。成果 报告的结果 模型部署时间 此前约一周，现在为几小时内 推理频率 每 30 分钟一次 成本影响 显著节省成本 工程工作 工艺工程师提出新的应用；机器学习工程师专注于流水线运行和监控 清理测试资源 对于测试部署，先禁用 Amazon EventBridge 规则并移除其目标。在正在运行的流水线执行和作业结束后，删除测试流水线及其 SageMaker 模型资源。如果为测试创建了实时端点，也将其删除。从 S3 中移除输入文件、中间文件和预测文件，并根据保留策略清理 CloudWatch 日志。删除模型文件和 Model Registry 版本之前，检查其他执行实例或审计要求是否仍依赖它们。结论 在 GS Caltex，工艺工程师在 SageMaker Canvas 中创建并验证模型，机器学习工程师使用这些资产在 SageMaker Pipelines 中运行生产工作流。这一交接减少了重复的集成工作。GS Caltex 报告称，机器学习工程师现在可在几小时内完成模型部署，而此前约需一周。除模型文件外，交接内容还需要包括预处理流程、输入模式、推理配置和审批流程。验证工作流能否复现预期结果，并按要求的时间表可靠运行。要尝试这种方法，可从一个已在 SageMaker Canvas 中验证的模型开始，并使用 GitHub 仓库。有关定义和运行工作流的详细信息，请参阅 SageMaker Pipelines 开发人员指南。通过手动执行检查预处理和批量预测结果，然后按应用所需的时间间隔安排运行。在第 2 部分，我们将介绍 GS Caltex 如何使用 Amazon Bedrock 自动将现有模型迁移到这条流水线。关于作者 WonGu Rho WonGu 是 GS Caltex DAX Lab 的数据分析师。他专长于数据分析、机器学习模型开发与运维、模型性能优化和 MLOps。此外，他通过设计和集成企业 AI 智能体、工作流和数据流水线，帮助业务团队提高运营效率并依据数据做出决策。Hyunsoo Kim 博士 Hyunsoo 是 Amazon Web Services 的生成式 AI 亚太及日本地区高级市场拓展专家解决方案架构师。他常驻首尔，帮助亚太地区的客户设计生成式 AI 解决方案，包括 AI 智能体、高级 RAG、模型优化和 MLOps。Sang Hyun Kim Sang 是 AWS Korea 的高级客户经理。他帮助能源和制造业客户采用云技术并将 AI 付诸实践。 他与 AWS 专家和合作伙伴携手，为客户提供从概念验证到生产部署及更广泛采用的支持。Joonhyung Kim Joonhyung 是 AWS 韩国的高级解决方案架构师，也是物理 AI 和空间计算专家。他帮助客户在 AWS 上构建和运营数字孪生、3D 和 AR/VR/XR 应用，重点关注交付速度和成本效率。他的经验涵盖生成式 AI、机器学习、数据和云架构。