用SageMaker流水线接管无代码模型部署

GS Caltex 把业务工程师在 Amazon SageMaker Canvas 里验证过的模型，接进了可以定时运行的生产流程。据公司报告，ML 工程师部署模型的时间从约一周缩短到几小时；推理每 30 分钟执行一次，也减少了常驻推理端点的需求。

这篇案例值得看的，是模型交付时容易漏掉的那半套东西：预处理、输入格式、推理配置、审批和调度。它也明确提醒，自动化接通之后，数据错位、任务重叠和完整成本仍要逐项验证。

从一周到几小时，省掉了哪些重复劳动

GS Caltex 是韩国能源与化工企业。他们让两类工程师各自负责擅长的部分：工艺工程师理解生产变量，在 SageMaker Canvas 中准备数据、训练模型，并判断预测有没有业务价值；ML 工程师负责部署、版本管理、定时执行和运行监控。

问题出在交接处。Canvas 能生成模型和 Data Wrangler 预处理流程，也能把模型登记到 Amazon SageMaker Model Registry，但这些资产还需要配置生产数据路径、推理容器和任务依赖，才能进入 SageMaker Pipelines。

他们自动化的是这段衔接，并复用工艺工程师已经验证过的资产。

这里要区分两条路径：发布新版本时，审核模型、确认配置并创建 SageMaker model resource；日常推理时，直接复用这个资源创建 SageMaker Batch Transform 作业。每 30 分钟跑一次预测，不需要每次重新注册、部署模型。

参考实现要求准备好 Canvas 模型及其 Data Wrangler 流程、CSV 推理数据、Amazon S3 bucket、SageMaker execution role、Amazon EventBridge role，以及安装了 Boto3 和 SageMaker Python SDK 2.x 的 Python 环境。它使用贷款分类示例，输入输出配置不能直接照搬到其他模型。

复用预处理，最容易错在输入契约

预处理流程不是一个可以随意换路径的黑盒。Data Wrangler 的 .flow 文件记录了数据源和变换步骤，生产环境必须保留已经验证的变换逻辑，同时正确替换数据来源及执行配置。

原文建议先在导出设置里关闭 Auto job configuration，选择 SageMaker Processing，再从实际 Processing job 中取得容器镜像 URI、名为 flow 的输入 S3 URI、执行角色及输出设置。不能假定每个 .flow 都包含 internal_metadata.dw_job.container_uri；跨 AWS Region 时，还要确认镜像是否可用。

执行角色需要读取输入、写入结果和访问容器镜像的权限。实例类型及数量则要按数据量选择。

另一个容易被忽略的差异是目标列。训练数据包含已知结果，推理数据没有，而训练阶段建立的流程可能仍要求这一列存在。参考实现采用五步处理：

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

要留意 DropTargetColumn：占位目标列只用于兼容预处理输入，进入模型之前必须移除。

占位值的类型和列位置都要匹配训练格式。这个办法也有明确边界：如果流程根据目标列过滤记录、派生特征，或执行目标编码，占位值可能改变其他特征。此时需要拆出推理所需的变换和训练阶段学到的状态，不能靠补一列解决。

模型输入是否正确，取决于列顺序、类型和变换结果共同匹配。

输出拼接也有类似风险。示例中的 AssembleResults 按行位置组合预处理数据和预测结果；一旦预处理排序、删行，或文件顺序发生变化，预测就可能对应到错误记录。生产中应检查文件顺序和行数，必要时维护源记录 ID 与预测结果的明确映射。

审批和流水线，都需要代码落实约束

Model Registry 可以把模型文件、推理配置和评估结果放在同一个 model package 中。仅拿到 model.tar.gz 还不够，它必须配合正确的推理容器和环境设置。

下面这段代码解决的是：部署前确认模型包已经完成且获批，再从该包创建供批量推理使用的模型资源。

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},
)

关键是检查 ModelApprovalStatus 的那一行。Registry 记录审批状态，却不会替每次推理请求强制执行审批检查。撤销批准之后，还需要更新使用该模型的调度和执行配置。示例为每个版本使用唯一 model_name，并把选定的模型包 ARN 留在流水线配置中。

五个步骤接入 Pipelines 时，处理器和 transformer 共用 PipelineSession。各处理器的 run() 返回步骤参数，分别交给 ProcessingStep；预测任务由 TransformStep 承接，transformer 指向已部署的 model_name。

步骤顺序通过 depends_on 控制，数据位置通过上一步的输出属性传递。例如 DropTargetColumn 定义名为 features 的输出，预测步骤读取其 S3 URI，使用 content_type="text/csv" 和 split_type="Line"。Data Wrangler 步骤则使用流程里定义的输出 ID，不能都写成 features。

结果组装还要接收预处理数据位置和 predict.properties.TransformOutput.S3OutputPath。流水线命名为 canvas-process-inference，通过 pipeline.upsert(role_arn=sagemaker_role_arn)["PipelineArn"] 取得 ARN，供后续调度使用。

每 30 分钟执行，并不保证便宜和准时

EventBridge 使用 put_rule 配置 ScheduleExpression="rate(30 minutes)"、State="ENABLED"，再用 put_targets 指向 pipeline_arn。目标角色 eventbridge_role_arn 必须信任 events.amazonaws.com，并拥有目标流水线的 sagemaker:StartPipelineExecution 权限；调用后还要检查 FailedEntryCount 和 FailedEntries。

rate(30 minutes) 表示每隔 30 分钟触发。若要求整点和半点执行，原文给出的是 cron(0/30 * * * ? *)。每次究竟处理哪个数据窗口，需要在输入准备阶段另行定义并记录。

30 分钟一次，每天就是 48 次计划运行。Pipelines 免去工作流服务器的维护，但 Processing 和 Batch Transform 仍使用各自的计算资源。估算成本要包含实例启动、预处理、推理、后处理，以及 S3、CloudWatch 等配套服务。

GS Caltex 报告推理成本明显降低，但原文没有给出金额、比例或对照负载。

运行时长同样要实测。流水线内部限制并行步骤，不能阻止两次独立执行重叠。参考实现虽然把中间和最终结果放进 runs/<execution-id>/，PrepareInput 却仍向 .flow 引用的固定 S3 位置写数据，共用这个位置的任务必须避免重叠。

要隔离输入，可以复制 .flow 并更新其中的数据源 URI。只声明 Pipelines 参数，不会自动替换文件内部的 URI。

我的判断

这套方案最有价值的部分，是把交付对象从模型文件扩展到了预处理、输入契约、推理配置和审批状态。对于已经使用 Canvas 的团队，这能减少在另一套代码里重写预处理所产生的偏差。

它的适用前提也很清楚：业务能接受批量预测，团队已经采用 AWS 相关服务，而且有人负责生产运行。需要稳定低延迟响应的请求，仍更适合 provisioned real-time endpoint；本来就用代码维护训练和推理流程的团队，也未必需要增加 Canvas 这一层。

部署从约一周缩短到几小时，是公司报告的结果。原文没有披露模型数量、复杂度及对照条件，不能据此预测一般团队能节省多少时间。成本下降也应视为该案例的结果，不能推出批量推理一定更便宜。

我更看重原文留下的限制条件：审批撤销如何生效、占位目标是否污染特征、结果按行拼接是否可靠。这些细节决定自动化能否用于生产。下一篇计划介绍用 Amazon Bedrock 自动迁移已有模型，本篇尚未给出实现，不能提前算作已验证能力。

能拿来做什么

• 如果正在把 Canvas 模型接入生产，先选一个已验证模型，整理 .flow、输入 schema、推理容器、模型包 ARN 和审批状态，完整手动运行一次，再考虑标准化交接。
• 给同一份验证数据分别跑 Canvas 和 Processing，比较列顺序、类型、行数及关键变换值，并确认占位目标没有影响特征；这些检查比仅确认任务成功更有用。
• 给每次执行记录数据窗口、预处理版本、模型包 ARN、流水线执行 ARN 和输出位置，同时验证重复窗口、任务重叠及错误输入会如何处理。
• 开启定时任务前，实测完整耗时和成本，确认预处理失败会阻止下游推理及结果发布，再根据测量结果配置 CloudWatch 失败与延迟告警。