1. 项目概述：把训练好的模型变成谁都能调用的“网络插座”

“Deploying Machine Learning Models as API using AWS”——这个标题听起来像一句技术说明书，但背后藏着一个非常现实的问题：你花了三周时间调参、清洗数据、跑通了XGBoost在客户流失预测上的AUC达到0.87，模型文件存进了S3，然后呢？业务部门问：“我们怎么把它接进CRM系统？”产品经理说：“能不能让销售APP点个按钮就返回预测分？”而运维同事默默看了你一眼，说：“别往我服务器上扔Python脚本。”

这就是模型部署（MLOps中真正的“Ops”环节）最常被低估的一环： 模型不是训练完就结束了，它必须以稳定、可监控、可伸缩、可鉴权的方式，暴露为一个HTTP端点 。AWS提供了一整套工具链来完成这件事，但它不是“一键部署”，而是一系列有明确取舍的技术决策组合——选EC2还是SageMaker？用API Gateway做路由还是直接用ALB？要不要加Lambda做预处理？模型版本怎么灰度？日志和指标往哪送？这些选择没有标准答案，只有场景适配。

我过去三年在金融、电商、IoT三个垂直领域落地过27个生产级ML API，其中19个跑在AWS上。踩过的坑比写的代码还多：比如某次用ECS部署TensorFlow Serving，因未限制内存导致OOM后容器反复重启，而CloudWatch告警阈值设在“连续5分钟无响应”，结果故障持续了42分钟才触发通知；又比如某次用SageMaker Endpoint自动扩缩，因未设置 InitialInstanceCount=1 ，冷启动延迟高达11秒，前端用户以为接口挂了直接刷新页面……这些都不是文档里会写的细节，但恰恰决定一个模型API是“能用”，还是“敢用”。

这篇文章不讲“如何创建IAM角色”，也不教“怎么写requirements.txt”。它聚焦在： 当你手头有一个 .pkl 、 .h5 或 saved_model ，想让它变成 https://api.yourcompany.com/v1/predict 时，AWS生态里最务实、最可控、最容易追查问题的那条路径是什么？每一步为什么这么选？参数背后藏着什么陷阱？ 适合已经跑通本地推理、正准备推到生产的算法工程师、MLOps工程师，也适合需要评估技术方案可行性的技术负责人。如果你还在用 flask run --host=0.0.0.0 --port=5000 测试模型，这篇就是为你写的。

---

2. 整体架构设计与方案选型逻辑

2.1 为什么不用“本地Flask+NGINX”直接上EC2？

这是新手最容易走的路：把训练好的模型加载进Flask，用Gunicorn起几个worker，再用NGINX反向代理，最后塞进一台t3.xlarge EC2。看起来简单，但实际生产中会立刻撞墙：

• 无健康检查集成 ：EC2本身不感知你的Flask进程是否真在服务请求。Gunicorn worker卡死、模型加载失败、甚至Python解释器崩溃，EC2实例状态仍是“running”，而负载均衡器（ALB/ELB）还在往它转发流量。
• 扩缩容形同虚设 ：你可以用Auto Scaling Group根据CPU使用率扩容EC2，但新实例启动→拉取代码→安装依赖→加载GB级模型→等待Flask ready，整个过程动辄3–5分钟。而真实业务流量尖峰（比如电商大促开场）往往在10秒内飙升300%，等你新机器起来，峰值早过去了。
• 版本管理混乱 ：模型更新意味着要SSH进机器、停服务、替换文件、重启进程。没有原子性，没有回滚点。某次我同事误删了 model_v2.pkl ，只能从S3恢复，期间17分钟服务不可用。
• 安全边界模糊 ：Flask默认不带认证、不强制HTTPS、日志混在stdout里难聚合。你要自己加JWT校验、配置ACM证书、写logrotate脚本——这些本该由平台层兜底的事，全堆在应用层。

所以， EC2裸跑Flask只适用于POC验证或日均请求<100次的内部工具 。一旦进入生产环境，就必须让基础设施承担它该承担的责任：健康探测、自动扩缩、蓝绿发布、集中鉴权、结构化日志。AWS的方案正是围绕这些能力构建的。

2.2 四种主流部署路径对比：SageMaker Endpoint、EC2+ECS、Lambda+API Gateway、Fargate

我们实测过四种组合在相同模型（ResNet-50图像分类，输入224x224 JPEG，输出top-3类别+置信度）下的关键指标：

方案	首字节延迟（P95）	冷启动（ms）	最大QPS（单实例）	模型更新耗时	运维复杂度	适用场景
SageMaker Endpoint	128 ms	0（常驻）	240	<2分钟（InService状态）	★★☆☆☆（低）	高SLA要求、需A/B测试、模型>500MB
EC2 + Docker + ALB	95 ms	0（常驻）	310	8–12分钟（Ansible滚动）	★★★★☆（高）	极致性能、定制化GPU驱动、遗留系统集成
Lambda + API Gateway	320 ms	1,800–3,500	15（内存6GB）	<1分钟（CodeDeploy）	★★☆☆☆（低）	轻量模型（<200MB）、事件驱动、突发流量、成本敏感
Fargate + ALB	165 ms	4,200–6,800	180	3–5分钟（ECS部署）	★★★☆☆（中）	无服务器但需容器、避免EC2管理、中等QPS

提示：冷启动指从无实例到首个请求返回的时间。Lambda冷启动包含下载代码、解压、初始化运行时、加载模型三阶段；Fargate冷启动包含拉取镜像、启动容器、执行ENTRYPOINT；SageMaker Endpoint无冷启动因其始终保活（但首次部署有“创建Endpoint”耗时）。

我们最终在80%的项目中选择了 SageMaker Endpoint ，原因很实在：

• 它原生支持 模型版本管理 （ ModelPackageGroup + ModelPackage ），一次注册，多次部署，回滚只需改Endpoint配置；
• 自动扩缩 基于 InvocationsPerInstance 指标（而非CPU），精准匹配模型吞吐瓶颈；
• 内置Data Capture 功能，可自动采样10%的请求/响应体存入S3，用于后续漂移检测；
• CloudWatch Metrics开箱即用 ： CPUUtilization , MemoryUtilization , ModelLatency , Invocation4XX , Invocation5XX 全部预置，无需自定义埋点；
• VPC内网直连 ：Endpoint可部署在私有子网，通过VPC Endpoint访问S3/Secrets Manager，避免公网暴露。

但SageMaker不是万能的。它的 最大约束是模型格式 ：必须打包为 tar.gz ，顶层目录含 code/ （inference.py）和 model/ （模型文件），且 inference.py 必须实现 model_fn , input_fn , predict_fn , output_fn 四个函数。这对PyTorch/TensorFlow原生用户友好，但对scikit-learn用户需额外封装——这也是本文后续重点展开的部分。

2.3 架构图：我们推荐的生产级最小可行架构

[Client] 
    ↓ HTTPS (TLS 1.2+)
[API Gateway (REST)] → 认证/限流/日志
    ↓ (Private Integration)
[VPC Link] 
    ↓ (Private DNS)
[SageMaker Endpoint] → 在私有子网中，Security Group仅允许ALB/VPC Link访问
    ↓ (异步)
[S3 Bucket] ← Data Capture（采样请求/响应）
    ↓
[CloudWatch Logs] ← inference.py中的print()自动捕获
[CloudWatch Metrics] ← SageMaker自动上报

关键设计点解析：

• API Gateway作为统一入口 ：不只是“转发”，它承担了身份校验（Cognito User Pool / IAM Authorizer）、请求限流（如500 req/min per API key）、结构化日志（Access Logging到S3）、错误映射（将SageMaker的500转为422）。这层剥离后，SageMaker Endpoint只需专注推理，逻辑更纯粹。
• VPC Link替代公网调用 ：SageMaker Endpoint默认分配公网DNS，但生产环境严禁模型服务暴露公网。VPC Link让API Gateway以私有方式调用Endpoint，全程不经过Internet，且延迟降低40%（实测从210ms→125ms）。
• S3 + CloudWatch双日志体系 ： print() 语句进CloudWatch Logs便于调试；Data Capture进S3用于长期分析。两者互补，缺一不可。

这个架构不是为了炫技，而是把每个组件的职责划清：API Gateway管“谁可以调”，VPC管“怎么安全调”，SageMaker管“怎么快准稳地算”，S3/CloudWatch管“调得怎么样”。责任分明，出问题时定位极快。

---

3. 核心细节解析与实操要点

3.1 SageMaker Endpoint的模型打包规范：绕不开的“四函数”契约

SageMaker不是让你丢个 .pkl 进去就完事。它要求你提供一个符合严格接口规范的 inference.py ，并打包进 model.tar.gz 。这个规范看似繁琐，实则是保障稳定性的基石——它强制你把模型加载、数据预处理、推理、后处理拆成独立阶段，每个阶段可单独测试、监控、超时控制。

inference.py 必须实现的四个函数：

# model/code/inference.py
import joblib
import numpy as np
import json
from io import BytesIO
from PIL import Image

# 1. model_fn: 加载模型（只执行一次，实例启动时）
def model_fn(model_dir):
    """
    model_dir: /opt/ml/model/ (SageMaker挂载的模型目录)
    返回：已加载的模型对象（会被缓存，供后续predict_fn复用）
    """
    # 注意：这里不能做heavy operation！如下载大文件、连接数据库
    model_path = f"{model_dir}/model.pkl"
    model = joblib.load(model_path)  # scikit-learn示例
    return model

# 2. input_fn: 解析请求体（每次请求都执行）
def input_fn(request_body, request_content_type):
    """
    request_body: bytes类型原始请求体
    request_content_type: 如 'application/json', 'image/jpeg'
    返回：预处理后的输入对象（如np.array, dict）
    """
    if request_content_type == 'application/json':
        data = json.loads(request_body.decode('utf-8'))
        # 假设JSON含{"features": [1.2, 3.4, ...]}
        return np.array(data['features']).reshape(1, -1)
    elif request_content_type == 'image/jpeg':
        # 图像处理：读取JPEG，转为RGB，resize，归一化
        img = Image.open(BytesIO(request_body)).convert('RGB').resize((224, 224))
        img_array = np.array(img) / 255.0  # 归一化到[0,1]
        return np.expand_dims(img_array, axis=0)  # 添加batch维度
    else:
        raise ValueError(f"Unsupported content type: {request_content_type}")

# 3. predict_fn: 执行推理（每次请求都执行）
def predict_fn(input_data, model):
    """
    input_data: input_fn返回的对象
    model: model_fn返回的对象
    返回：原始预测结果（如np.ndarray, list）
    """
    # 关键：此处应尽量轻量，避免I/O、网络调用
    # 如果模型本身支持batch，这里传入batch数据
    prediction = model.predict(input_data)
    return prediction

# 4. output_fn: 格式化响应（每次请求都执行）
def output_fn(prediction, response_content_type):
    """
    prediction: predict_fn返回的结果
    response_content_type: 如 'application/json'
    返回：bytes类型响应体
    """
    if response_content_type == 'application/json':
        # 将numpy array转为list，再json序列化
        result = {
            "prediction": prediction.tolist() if hasattr(prediction, 'tolist') else prediction,
            "timestamp": int(time.time())
        }
        return json.dumps(result).encode('utf-8')
    else:
        raise ValueError(f"Unsupported accept type: {response_content_type}")

注意： model_fn 中禁止任何网络I/O（如 requests.get ）、大文件下载、数据库连接。因为SageMaker会在实例启动时调用它一次，若超时（默认15秒）则实例启动失败。所有外部依赖必须在 input_fn 或 predict_fn 中按需获取，并做好重试/降级。

实操心得：

• 模型文件命名必须明确 ：不要用 model.joblib 这种模糊名，用 rf_classifier_v2_20240515.pkl ，版本和日期一目了然，方便Data Capture分析时关联。
• input_fn 是安全阀 ：在这里做输入校验。例如检查 len(features)==128 ，若不满足立即 raise ValueError("Invalid feature dimension") ，避免无效请求进入 predict_fn 消耗GPU资源。
• output_fn 决定客户端体验 ：返回JSON时，务必用 prediction.tolist() 而非 str(prediction) ，否则NumPy数组会变成 [[1.23456789]] 这种不可读字符串，前端解析失败。

3.2 Docker容器化部署（EC2/ECS/Fargate）：当SageMaker不适用时的备选

某些场景SageMaker无法满足：

• 模型依赖特定CUDA/cuDNN版本（如TensorRT优化模型）；
• 需要调用C++共享库（ .so ）；
• 必须复用现有Kubernetes Operator；
• 合规要求模型进程必须运行在物理隔离的EC2上。

此时，Docker是更灵活的选择。核心原则： 容器镜像必须是“自包含”的，不依赖启动时下载任何东西 。

我们采用的标准Dockerfile结构：

# 使用Amazon Linux 2官方基础镜像（与EC2/ECS兼容性最好）
FROM amazon/aws-cli:2.13.12

# 安装必要系统依赖
RUN yum update -y && \
    yum install -y python39 python39-pip python39-devel gcc gcc-c++ && \
    yum clean all

# 创建非root用户（安全强制要求）
RUN groupadd -g 1001 -f appuser && \
    useradd -r -u 1001 -g appuser appuser

# 复制并安装Python依赖（注意：requirements.txt必须锁定版本）
COPY requirements.txt .
RUN pip3.9 install --no-cache-dir -r requirements.txt

# 复制应用代码和模型（模型文件建议放S3，此处仅为演示）
COPY app/ /app/
WORKDIR /app

# 切换到非root用户
USER appuser

# 暴露端口（必须与应用监听端口一致）
EXPOSE 8080

# 启动命令：Gunicorn + Uvicorn（比纯Flask更健壮）
CMD exec gunicorn --bind :8080 --workers 4 --worker-class uvicorn.workers.UvicornWorker --timeout 120 --keep-alive 5 app:app

app.py 核心逻辑（Uvicorn + FastAPI）：

from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
import joblib
import numpy as np
import time

app = FastAPI()

# 全局模型缓存（单例模式）
_model = None

@app.on_event("startup")
async def load_model():
    global _model
    start = time.time()
    try:
        _model = joblib.load("/app/model/model.pkl")
        print(f"[INFO] Model loaded in {time.time()-start:.2f}s")
    except Exception as e:
        print(f"[ERROR] Failed to load model: {e}")
        raise

class PredictRequest(BaseModel):
    features: list[float]

class PredictResponse(BaseModel):
    prediction: float
    latency_ms: float

@app.post("/v1/predict", response_model=PredictResponse)
async def predict(request: PredictRequest):
    if _model is None:
        raise HTTPException(status_code=503, detail="Model not loaded")
    
    start = time.time()
    try:
        # 输入校验
        if len(request.features) != 128:
            raise HTTPException(status_code=400, detail="Feature dimension must be 128")
        
        # 推理
        X = np.array(request.features).reshape(1, -1)
        pred = _model.predict(X)[0]
        
        return PredictResponse(
            prediction=float(pred),
            latency_ms=(time.time() - start) * 1000
        )
    except Exception as e:
        print(f"[ERROR] Prediction failed: {e}")
        raise HTTPException(status_code=500, detail="Prediction error")

关键细节：

• @app.on_event("startup") 确保模型在Gunicorn worker启动时加载 ，而非每次请求加载，避免重复I/O。
• Depends 和 BaseModel 提供自动请求校验 ，比手动 if 判断更可靠。
• latency_ms 返回给客户端 ，方便前端监控自身调用质量，形成闭环。

部署到ECS时，Task Definition中必须设置：

• memoryReservation: 4096 （预留4GB内存，防止OOM）
• healthCheck : "CMD-SHELL","curl -f http://localhost:8080/health || exit 1" （健康检查端点）
• logConfiguration : "logDriver":"awslogs" ，指向CloudWatch Log Group

实操心得：我们曾因忘记在ECS Task中设置 healthCheck ，导致ALB持续向一个已卡死的容器转发流量。后来强制要求： 所有生产ECS Task必须配置健康检查，且路径 /health 返回 {"status":"ok","uptime":12345} ，不包含任何业务逻辑 。

3.3 API Gateway集成：不止是“转发”，更是API治理中枢

很多人把API Gateway当成“高级Nginx”，只配个 integration.request.uri 。这是巨大浪费。它真正的价值在于 将模型API纳入企业级API治理体系 。

我们必配的五项设置：

1. Cognito User Pool Authorizer
   创建Cognito User Pool，添加App Client，然后在API Gateway中创建Authorizer，选择该User Pool。在Method Request中启用此Authorizer。这样，每个请求Header必须带 Authorization: Bearer <id_token> ，Gateway自动校验签名、过期时间、scope，非法Token直接返回401， 模型服务完全不接触未授权请求 。

2. Usage Plan + API Key
   创建Usage Plan，绑定API Stage（如 prod ），设置Rate Limit（如100 req/sec）和Quota（如10000 req/day）。为每个调用方（如CRM系统、APP）生成独立API Key，并关联到该Plan。Key泄露时，可单独禁用，不影响其他系统。

3. Request Validator
   启用 Validate request body 和 Validate request parameters 。定义JSON Schema强制校验请求体结构。例如：

   {
     "type": "object",
     "properties": {
       "features": {
         "type": "array",
         "items": {"type": "number"},
         "minItems": 128,
         "maxItems": 128
       }
     },
     "required": ["features"]
   }
   

   不符合Schema的请求，Gateway在转发前就返回400， 保护后端免受畸形请求冲击 。

4. Access Logging to S3
   开启Access Logging，日志格式包含 $context.identity.sourceIp , $context.httpMethod , $context.status , $context.integrationLatency , $context.error.message 。日志存入S3后，可用Athena做SQL分析，例如：“找出所有503错误中， integrationLatency > 10000 的请求，关联其IP和UserAgent”。

5. Custom Domain Name + ACM Certificate
   绑定 api.yourcompany.com ，使用ACM签发的证书。 绝对禁止使用 xxx.execute-api.region.amazonaws.com 这种默认域名 ——它暴露AWS账户ID，且无法做品牌统一。

提示：API Gateway的 Integration Timeout 默认29秒，但SageMaker Endpoint的 MaxConcurrentInvocationsPerInstance 可能让请求排队。我们将其设为 25 秒，并在 output_fn 中加入 try/except 捕获 TimeoutError ，返回友好的 {"error":"timeout", "retry_after": "5"} ，引导客户端指数退避重试。

---

4. 实操过程与核心环节实现

4.1 从本地模型到SageMaker Endpoint的完整流水线

假设你本地有一个训练好的scikit-learn模型 churn_model_v3.pkl ，现在要部署为 https://api.yourcompany.com/v1/churn 。

Step 1：准备模型包（model.tar.gz）

# 创建目录结构
mkdir -p model/code model/model

# 复制模型文件
cp churn_model_v3.pkl model/model/

# 编写inference.py（见3.1节）
cat > model/code/inference.py << 'EOF'
import joblib
import numpy as np
import json
import time

def model_fn(model_dir):
    return joblib.load(f"{model_dir}/model.pkl")

def input_fn(request_body, request_content_type):
    if request_content_type == 'application/json':
        data = json.loads(request_body.decode('utf-8'))
        return np.array(data['features']).reshape(1, -1)
    else:
        raise ValueError(f"Unsupported content type: {request_content_type}")

def predict_fn(input_data, model):
    start = time.time()
    pred = model.predict_proba(input_data)[0, 1]  # 返回流失概率
    print(f"[DEBUG] Inference time: {time.time()-start:.3f}s")
    return pred

def output_fn(prediction, response_content_type):
    if response_content_type == 'application/json':
        return json.dumps({
            "churn_probability": float(prediction),
            "risk_level": "high" if prediction > 0.7 else "medium" if prediction > 0.3 else "low"
        }).encode('utf-8')
    else:
        raise ValueError(f"Unsupported accept type: {response_content_type}")
EOF

# 打包
cd model
tar -czf ../model.tar.gz .
cd ..

Step 2：上传模型包至S3

# 创建专用S3桶（开启版本控制+服务器端加密）
aws s3 mb s3://yourcompany-ml-models --region us-east-1
aws s3api put-bucket-versioning --bucket yourcompany-ml-models --versioning-configuration Status=Enabled
aws s3api put-bucket-encryption --bucket yourcompany-ml-models \
  --server-side-encryption-configuration '{"Rules":[{"ApplyServerSideEncryptionByDefault":{"SSEAlgorithm":"AES256"}}]}'

# 上传
aws s3 cp model.tar.gz s3://yourcompany-ml-models/churn/v3/model.tar.gz

Step 3：创建SageMaker Model（逻辑实体）

import boto3
import sagemaker
from sagemaker.sklearn.model import SKLearnModel

sagemaker_session = sagemaker.Session()
role = sagemaker.get_execution_role()  # IAM Role with SageMakerFullAccess

model = SKLearnModel(
    model_data='s3://yourcompany-ml-models/churn/v3/model.tar.gz',
    role=role,
    framework_version='0.23-1',  # scikit-learn version
    py_version='py3',
    entry_point='inference.py',
    source_dir='model/code',
    sagemaker_session=sagemaker_session
)

# 创建Model（此操作注册模型，不启动实例）
model.create(
    model_name='churn-model-v3',
    instance_type='ml.m5.xlarge'  # 仅用于创建，不启动
)

Step 4：部署Endpoint（启动实例）

# 配置自动扩缩策略
predictor = model.deploy(
    initial_instance_count=1,
    instance_type='ml.m5.xlarge',
    endpoint_name='churn-endpoint-prod',
    # 自动扩缩：当每实例每秒请求数 > 10 时扩容，< 5 时缩容
    auto_scaling_enabled=True,
    min_instance_count=1,
    max_instance_count=4,
    scaling_target_invocations_per_instance=10,
    # CloudWatch Alarms名称前缀
    tags=[{'Key': 'Project', 'Value': 'churn'}]
)

print(f"Endpoint deployed: {predictor.endpoint}")

Step 5：API Gateway集成（CloudFormation模板片段）

Resources:
  ChurnApi:
    Type: AWS::ApiGateway::RestApi
    Properties:
      Name: churn-api-prod
      Description: Churn prediction API
      EndpointConfiguration:
        Types: [PRIVATE]  # 关键：设为PRIVATE
      Policy: |
        {
          "Version": "2012-10-17",
          "Statement": [
            {
              "Effect": "Allow",
              "Principal": "*",
              "Action": "execute-api:Invoke",
              "Resource": "arn:aws:execute-api:*:*:*/*/*/*"
            }
          ]
        }

  VpcLink:
    Type: AWS::ApiGateway::VpcLink
    Properties:
      Name: churn-vpclink
      Description: Link to SageMaker VPC
      TargetArns:
        - !Sub 'arn:aws:elasticloadbalancing:${AWS::Region}:${AWS::AccountId}:loadbalancer/app/sagemaker-endpoint-churn/${VpcLinkId}'

  ChurnMethod:
    Type: AWS::ApiGateway::Method
    Properties:
      RestApiId: !Ref ChurnApi
      ResourceId: !GetAtt ChurnApi.RootResourceId
      HttpMethod: POST
      AuthorizationType: COGNITO_USER_POOLS
      AuthorizerId: !Ref CognitoAuthorizer
      Integration:
        Type: HTTP_PROXY
        IntegrationHttpMethod: POST
        Uri: !Sub 'http://${SageMakerEndpointDnsName}/invocations'
        RequestParameters:
          integration.request.header.Content-Type: "'application/json'"
        # VPC Link集成
        ConnectionType: VPC_LINK
        ConnectionId: !Ref VpcLink

Step 6：测试与验证

# 获取Endpoint DNS（SageMaker控制台或boto3 describe_endpoint）
ENDPOINT_URL="https://runtime.sagemaker.us-east-1.amazonaws.com/endpoints/churn-endpoint-prod/invocations"

# 直接调用（测试SageMaker层）
curl -X POST $ENDPOINT_URL \
  -H "Content-Type: application/json" \
  -d '{"features": [0.2, 0.8, 1.1, ...]}' \
  --aws-sigv4 "aws:amz:us-east-1:sagemaker" \
  --user "$AWS_ACCESS_KEY:$AWS_SECRET_KEY"

# 通过API Gateway调用（生产路径）
curl -X POST https://api.yourcompany.com/v1/churn \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer <valid-jwt-token>" \
  -H "x-api-key: <your-api-key>" \
  -d '{"features": [0.2, 0.8, 1.1, ...]}'

实操心得：第一次部署时，务必在CloudWatch Logs中查看 /aws/sagemaker/Endpoints/churn-endpoint-prod 日志流。如果看到 ImportError: No module named 'joblib' ，说明 requirements.txt 没打进 code/ 目录或 entry_point 路径错了。我们有个检查清单：① model.tar.gz 解压后 code/ 下有 inference.py 和 requirements.txt ；② requirements.txt 第一行是 joblib==1.2.0 ；③ inference.py 中 import 语句无拼写错误。

4.2 生产环境必备监控与告警配置

部署完成只是开始。真正的挑战是让API“可观察”。

CloudWatch Metrics（SageMaker自动上报）

• Invocations : 总请求数（按Endpoint、Variant、Instance维度）
• ModelLatency : 从收到请求到返回响应的毫秒数（P50/P90/P95/P99）
• OverheadLatency : SageMaker框架开销（模型加载、序列化等，应<50ms）
• CPUUtilization / MemoryUtilization : 实例资源水位
• Invocation4XX / Invocation5XX : 客户端错误/服务端错误计数

我们创建的告警规则（CloudWatch Alarms）：

Alarm Name	Metric	Condition	Threshold	Actions	说明
Churn-Endpoint-5XX-High	Invocation5XX	>= 5 for 1 minute	5	SNS Topic (PagerDuty)	表明模型或代码异常，需立即介入
Churn-Endpoint-Latency-P95	ModelLatency	>= 1000 for 5 minutes	1000ms	SNS Topic (Slack)	P95延迟超标，可能模型变慢或实例过载
Churn-Endpoint-CPU-High	CPUUtilization	>= 85 for 10 minutes	85%	Auto Scaling: Increase Instance Count	CPU成为瓶颈，需扩容
Churn-Endpoint-Invocations-Spike	Invocations	Increase > 200% from 1 hour ago	—	SNS Topic (Team)	流量突增，需确认是否正常活动

Data Capture配置（S3采样）

在Endpoint配置中启用：

predictor.update_endpoint_weights_and_capacities(
    EndpointName='churn-endpoint-prod',
    DesiredWeightAndCapacities=[
        {
            'VariantName': 'AllTraffic',
            'DesiredWeight': 100.0,
            'DesiredInstanceCount': 1
        }
    ]
)

# 启用Data Capture（采样10%）
predictor.enable_data_capture(
    destination_s3_uri='s3://yourcompany-ml-data-capture/churn/',
    kms_key_id='arn:aws:kms:us-east-1:123456789012:key/abcd1234-...',
    sampling_percentage=10
)

采样数据结构（S3中JSONL文件）：

{
  "event_version": "1.0",
  "source": "aws.sagemaker",
  "account": "123456789012",
  "time": "2024-05-15T08:23:45Z",
  "region": "us-east-1",
  "resources": [{"ARN": "arn:aws:sagemaker:us-east-1:123456789012:endpoint/churn-endpoint-prod"}],
  "detail": {
    "endpointName": "churn-endpoint-prod",
    "variantName": "AllTraffic",
    "modelName": "churn-model-v3",
    "contentType": "application/json",
    "accept": "application/json",
    "invocationStartTime": "2024-05-15T08:23:45.123Z",
    "invocationEndTime": "2024-05-15T08:23:45.456Z",
    "payload": "{\"features\": [0.2, 0.8, ...]}",
    "response": "{\"churn_probability\": 0.67, \"risk_level\": \"medium\"}"
  }
}

实操心得：Data Capture的S3桶必须开启 生命周期策略 ，自动将30天前的对象转为Glacier，否则存储费用会失控。我们曾因忘记配置，在6个月后发现S3账单激增$2,300/月。

---

5. 常见问题与排查技巧实录

5.1 “Endpoint状态卡在Creating，30分钟后失败”

现象 ：调用 create_endpoint 后， describe_endpoint 返回 Status: Creating ，30分钟后变为 Failed ， FailureReason 显示 Unable to create VPC endpoint 或 ResourceLimitExceeded 。

排查路径 ：

1. 检查VPC Endpoint Service ：SageMaker需要 com.amazonaws.<region>.sagemaker.runtime 这个VPC Endpoint。进入VPC控制台 → Endpoints → Create Endpoint → 选择 com.amazonaws.us-east-1.sagemaker.runtime 。若列表中无此选项，说明你的AWS账户在该区域未启用SageMaker服务（需联系AWS Support开通）。
2. 检查Security Group入站规则 ：Endpoint所在子网的Security Group，必须允许来自ALB或VPC Link的 TCP:8080 （或你指定的端口）入站。常见错误是只开了 0.0.0.0/0 ，但生产环境应精确到ALB的安全组ID。
3. 检查IAM权限 ：执行 create_endpoint 的IAM Role，必须有 sagemaker:CreateEndpoint 、 ec2:CreateNetworkInterface 、 ec2:DescribeSubnets 、 ec2:DescribeSecurityGroups 权限。最小权限策略见AWS官方文档 AmazonSageMakerFullAccess 的子集。

根本解决 ：我们编写了一个预检脚本（ precheck.py ），在 deploy 前自动运行：

def precheck_vpc():
    ec2 = boto3.client('ec2', region_name='us-east-1')
    # 检查子网是否存在且状态ok
    subnets = ec2.describe_subnets(SubnetIds=['subnet-12345'])['Subnets']
    assert subnets[0]['State