AI 工程师云计算指南：用 AWS/GCP/Azure 构建 AI 服务

1. 云 AI 平台概览
2. AWS AI/ML 服务
3. GCP AI/ML 服务
4. Azure AI Services
- Azure Machine Learning
  - 运行 Azure ML 训练作业
- Azure OpenAI Service
5. Kubernetes for AI（EKS/GKE/AKS）
6. 无服务器 AI 推理
- 冷启动优化
- 用 AWS Fargate 实现基于容器的无服务器架构
7. 面向 AI 的数据存储
8. 云 AI 监控
- 模型性能漂移检测
- CloudWatch 自定义指标告警
9. MLOps on Cloud
10. 测验
参考资料

1. 云 AI 平台概览

对 AI 工程师而言，云计算如今已经不是可选项，而是必需品。能够按需即时置备数百个 GPU，把从模型训练到服务上线的整个 ML 生命周期都交给托管服务处理——这样的环境已经齐备。

IaaS、PaaS、SaaS 概念

云服务分为三个层次。

IaaS（Infrastructure as a Service）：提供虚拟机、存储、网络。EC2 GPU 实例是典型代表。拥有最大控制权，但基础设施管理负担也最重。
PaaS（Platform as a Service）：连运行时和中间件都替你管理好。AWS SageMaker、GCP Vertex AI、Azure ML 属于这一类，可以专注于模型代码本身。
SaaS（Software as a Service）：把完成的 AI 功能以 API 形式提供出来。AWS Bedrock、GCP Gemini API、Azure OpenAI Service 是代表性例子。

主要云 AI 服务对比

功能	AWS	GCP	Azure
ML 平台	SageMaker	Vertex AI	Azure ML
LLM API	Bedrock	Vertex AI (Gemini)	Azure OpenAI
托管 Notebook	SageMaker Studio	Vertex AI Workbench	Azure ML Studio
AutoML	SageMaker Autopilot	Vertex AutoML	Azure AutoML
Feature Store	SageMaker Feature Store	Vertex Feature Store	Azure ML Feature Store
模型注册表	SageMaker Model Registry	Vertex Model Registry	Azure ML Registry
无服务器推理	Lambda, Fargate	Cloud Run, Cloud Functions	Azure Functions, Container Apps

GPU 实例类型对比

AWS GPU 实例：

p3.2xlarge：V100 x1，61 GiB 内存 — 小规模训练
p4d.24xlarge：A100 x8，320 GiB 内存 — 大规模分布式训练
p5.48xlarge：H100 x8，2 TiB 内存 — 最新 LLM 训练

GCP GPU 实例：

n1-standard-8 + V100：成本效益高的训练
a2-highgpu-8g：A100 x8 — Vertex AI 默认训练配置
a3-highgpu-8g：H100 x8 — 最新大型模型

Azure GPU 实例：

NC6s_v3：V100 x1 — 开发/测试
ND96asr_v4：A100 x8 — 大规模训练
ND96amsr_A100_v4：A100 80GB x8 — 最高性能

成本优化策略

云 AI 成本中有 70% 以上来自计算资源。主要的节省策略如下。

Spot/Preemptible 实例：相比 On-Demand 最多可节省 90%，必须针对中断做好检查点（checkpointing）。
预留实例 / Committed Use：以 1~3 年的承诺换取 40~60% 的节省，适合长期项目。
自动扩缩容（Auto Scaling）：根据推理流量自动调整实例数量。
Savings Plans（AWS）：以计算用量承诺换取灵活的实例类型折扣。

2. AWS AI/ML 服务

SageMaker 核心功能

Amazon SageMaker 是 AWS 的一体化 ML 平台。从数据准备到模型部署、监控，整个 ML 生命周期都在同一个服务里完成。

import sagemaker
from sagemaker.pytorch import PyTorch
from sagemaker import get_execution_role

role = get_execution_role()
sess = sagemaker.Session()

# SageMaker Training Job
estimator = PyTorch(
    entry_point='train.py',
    source_dir='./src',
    role=role,
    instance_type='ml.p4d.24xlarge',
    instance_count=4,
    framework_version='2.1.0',
    py_version='py310',
    hyperparameters={
        'epochs': 10,
        'batch-size': 32,
        'learning-rate': 0.001
    },
    distribution={
        'torch_distributed': {'enabled': True}
    }
)

estimator.fit({'train': 's3://bucket/train', 'val': 's3://bucket/val'})

通过 SageMaker 的 distribution 参数，可以轻松设置基于 PyTorch DDP 的分布式训练。只要同时指定 instance_count=4 与 torch_distributed 选项，就会自动配置跨 4 个节点的数据并行训练。

SageMaker 模型部署

from sagemaker.pytorch import PyTorchModel

model = PyTorchModel(
    model_data='s3://bucket/model.tar.gz',
    role=role,
    framework_version='2.1.0',
    py_version='py310',
    entry_point='inference.py'
)

predictor = model.deploy(
    initial_instance_count=2,
    instance_type='ml.g4dn.xlarge',
    endpoint_name='my-pytorch-endpoint'
)

# 调用预测端点
result = predictor.predict({'inputs': 'Hello, cloud AI!'})

AWS Bedrock（LLM API）

AWS Bedrock 是可以用单一 API 调用 Anthropic Claude、Meta Llama、Amazon Titan 等多种基础模型的服务。

import boto3
import json

bedrock = boto3.client('bedrock-runtime', region_name='us-east-1')

response = bedrock.invoke_model(
    modelId='anthropic.claude-3-5-sonnet-20241022-v2:0',
    body=json.dumps({
        'anthropic_version': 'bedrock-2023-05-31',
        'max_tokens': 1024,
        'messages': [{'role': 'user', 'content': '给我讲讲最近的 AI 趋势'}]
    })
)
result = json.loads(response['body'].read())
print(result['content'][0]['text'])

用 AWS Lambda 实现无服务器推理

轻量模型可以用 Lambda 做无服务器部署。

import json
import boto3
import numpy as np

def lambda_handler(event, context):
    # 解析事件中的输入数据
    body = json.loads(event['body'])
    input_data = body['input']

    # 从 S3 加载模型（建议打包进容器镜像）
    prediction = run_inference(input_data)

    return {
        'statusCode': 200,
        'body': json.dumps({'prediction': prediction})
    }

3. GCP AI/ML 服务

Vertex AI Training

Google Cloud 的 Vertex AI 是一体化 ML 平台，与 BigQuery 的深度集成是它的强项。

from google.cloud import aiplatform

aiplatform.init(project='my-project', location='us-central1')

job = aiplatform.CustomTrainingJob(
    display_name='pytorch-training',
    script_path='train.py',
    container_uri='us-docker.pkg.dev/vertex-ai/training/pytorch-gpu.2-0:latest',
    requirements=['transformers', 'datasets']
)

model = job.run(
    dataset=None,
    machine_type='a2-highgpu-8g',
    accelerator_type='NVIDIA_TESLA_A100',
    accelerator_count=8,
    args=['--epochs=10', '--batch_size=32']
)

Vertex AI 模型部署

from google.cloud import aiplatform

# 上传模型
model = aiplatform.Model.upload(
    display_name='my-pytorch-model',
    artifact_uri='gs://bucket/model/',
    serving_container_image_uri='us-docker.pkg.dev/vertex-ai/prediction/pytorch-gpu.2-0:latest'
)

# 创建端点并部署
endpoint = aiplatform.Endpoint.create(display_name='my-endpoint')
model.deploy(
    endpoint=endpoint,
    dedicated_resources_machine_type='n1-standard-4',
    dedicated_resources_accelerator_type='NVIDIA_TESLA_T4',
    dedicated_resources_accelerator_count=1,
    min_replica_count=1,
    max_replica_count=5
)

BigQuery ML

BigQuery ML 是用 SQL 语法就能训练与预测 ML 模型的强大工具。

-- 用 BigQuery ML 训练分类模型
CREATE OR REPLACE MODEL `dataset.fraud_model`
OPTIONS(
    model_type = 'BOOSTED_TREE_CLASSIFIER',
    num_parallel_tree = 1,
    max_iterations = 50,
    input_label_cols = ['is_fraud']
) AS
SELECT * FROM `dataset.transactions_train`;

-- 评估模型
SELECT *
FROM ML.EVALUATE(MODEL `dataset.fraud_model`,
  (SELECT * FROM `dataset.transactions_test`)
);

-- 执行预测
SELECT *
FROM ML.PREDICT(MODEL `dataset.fraud_model`,
  (SELECT * FROM `dataset.new_transactions`)
);

4. Azure AI Services

Azure Machine Learning

Azure ML 是微软的企业级 ML 平台，与 Active Directory 的集成以及混合云支持是它的强项。

from azure.ai.ml import MLClient
from azure.ai.ml.entities import AmlCompute, Command
from azure.identity import DefaultAzureCredential

ml_client = MLClient(
    DefaultAzureCredential(),
    subscription_id="YOUR_SUBSCRIPTION",
    resource_group_name="rg-ai",
    workspace_name="ai-workspace"
)

# 创建 GPU 集群
compute_config = AmlCompute(
    name="gpu-cluster",
    type="amlcompute",
    size="Standard_ND96asr_v4",
    min_instances=0,
    max_instances=4,
    idle_time_before_scale_down=120
)
ml_client.compute.begin_create_or_update(compute_config).result()

运行 Azure ML 训练作业

from azure.ai.ml.entities import Command
from azure.ai.ml import Input

job = Command(
    code="./src",
    command="python train.py --epochs 10 --learning_rate 0.001",
    environment="AzureML-pytorch-2.0-ubuntu20.04-py38-cuda11-gpu@latest",
    compute="gpu-cluster",
    inputs={
        "train_data": Input(type="uri_folder", path="azureml://datastores/mydata/paths/train/")
    },
    display_name="pytorch-training-job"
)

returned_job = ml_client.jobs.create_or_update(job)
print(f"Job URL: {returned_job.studio_url}")

Azure OpenAI Service

from openai import AzureOpenAI

client = AzureOpenAI(
    azure_endpoint="https://your-resource.openai.azure.com/",
    api_key="YOUR_API_KEY",
    api_version="2024-02-01"
)

response = client.chat.completions.create(
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "You are an AI assistant."},
        {"role": "user", "content": "给我讲讲云 AI 的优点"}
    ]
)
print(response.choices[0].message.content)

5. Kubernetes for AI（EKS/GKE/AKS）

Kubernetes 已成为大规模 AI 工作负载编排的标准。

GPU 节点池配置

# GKE GPU 节点池（Terraform）
resource "google_container_node_pool" "gpu_pool" {
name       = "gpu-pool"
cluster    = google_container_cluster.primary.name
node_count = 2

node_config {
machine_type = "a2-highgpu-1g"
guest_accelerator {
type  = "nvidia-tesla-a100"
count = 1
}
oauth_scopes = ["https://www.googleapis.com/auth/cloud-platform"]
}
}

NVIDIA 设备插件

kubectl apply -f https://raw.githubusercontent.com/NVIDIA/k8s-device-plugin/v0.14.1/nvidia-device-plugin.yml

Kubeflow Pipeline 定义

import kfp
from kfp import dsl

@dsl.component(
    base_image='python:3.10',
    packages_to_install=['scikit-learn', 'pandas']
)
def train_model(
    data_path: str,
    model_path: kfp.dsl.OutputPath(str)
):
    import pickle
    from sklearn.ensemble import RandomForestClassifier
    import pandas as pd

    df = pd.read_csv(data_path)
    X = df.drop('label', axis=1)
    y = df['label']

    model = RandomForestClassifier(n_estimators=100)
    model.fit(X, y)

    with open(model_path, 'wb') as f:
        pickle.dump(model, f)

@dsl.pipeline(name='ml-pipeline')
def ml_pipeline(data_path: str = 'gs://bucket/data.csv'):
    train_task = train_model(data_path=data_path)

用 KEDA 实现自动扩缩容

KEDA（Kubernetes Event-driven Autoscaling）根据队列深度或 HTTP 请求数，自动扩缩 AI 推理 Pod。

apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: inference-scaler
spec:
  scaleTargetRef:
    name: inference-deployment
  minReplicaCount: 1
  maxReplicaCount: 20
  triggers:
    - type: aws-sqs-queue
      metadata:
        queueURL: https://sqs.us-east-1.amazonaws.com/123456789/inference-queue
        queueLength: '5'

6. 无服务器 AI 推理

对于流量间歇性或不可预测的 AI 服务而言，无服务器是一个划算的选择。

冷启动优化

ML 模型的冷启动可能耗时数秒到数十秒。最小化的方法：

预置并发（AWS Lambda）：提前维持已预热的实例
容器镜像优化：移除不必要的软件包，使用多阶段构建
模型量化：用 FP16/INT8 把模型体积压缩到一半以下
延迟加载：在处理函数外部初始化模型（利用全局变量）

# Lambda 冷启动优化模式
import boto3
import json
from transformers import pipeline

# 用全局变量加载模型（处理函数外部）
# 容器被复用时，这段代码不会再次执行
classifier = pipeline('sentiment-analysis', model='distilbert-base-uncased-finetuned-sst-2-english')

def lambda_handler(event, context):
    body = json.loads(event['body'])
    text = body['text']

    result = classifier(text)

    return {
        'statusCode': 200,
        'body': json.dumps(result)
    }

用 AWS Fargate 实现基于容器的无服务器架构

Fargate 无需管理服务器即可运行容器。相比 Lambda，没有内存/时间限制，可以承载大体量模型的服务。

{
  "family": "ai-inference-task",
  "networkMode": "awsvpc",
  "requiresCompatibilities": ["FARGATE"],
  "cpu": "4096",
  "memory": "16384",
  "containerDefinitions": [
    {
      "name": "inference-container",
      "image": "123456789.dkr.ecr.us-east-1.amazonaws.com/my-model:latest",
      "portMappings": [{ "containerPort": 8080 }],
      "environment": [{ "name": "MODEL_PATH", "value": "/opt/ml/model" }]
    }
  ]
}

7. 面向 AI 的数据存储

对象存储对比

服务	提供商	特点
Amazon S3	AWS	11 个 9 的持久性，SDK 生态丰富
Google Cloud Storage	GCP	与 BigQuery 原生集成
Azure Blob Storage	Azure	支持 Azure Data Lake Gen2

数据湖架构

面向 AI 数据流水线的高效 Lakehouse 模式：

Raw Layer（Bronze）
    └── 原样保留原始数据
    └── 分区：year/month/day/

Processed Layer（Silver）
    └── 清洗、去重、强制 Schema
    └── Parquet/Delta Lake 格式

Feature Layer（Gold）
    └── ML 特征工程已完成
    └── 已注册到 Feature Store

大模型检查点管理

import boto3
from pathlib import Path

def save_checkpoint_to_s3(model, optimizer, epoch, loss, bucket, prefix):
    """把模型检查点保存到 S3"""
    import torch
    checkpoint = {
        'epoch': epoch,
        'model_state_dict': model.state_dict(),
        'optimizer_state_dict': optimizer.state_dict(),
        'loss': loss,
    }
    local_path = f'/tmp/checkpoint_epoch_{epoch}.pt'
    torch.save(checkpoint, local_path)

    s3 = boto3.client('s3')
    s3_key = f'{prefix}/checkpoint_epoch_{epoch}.pt'
    s3.upload_file(local_path, bucket, s3_key)
    print(f'Checkpoint saved to s3://{bucket}/{s3_key}')

8. 云 AI 监控

模型性能漂移检测

生产模型的性能可能随时间推移而下降。SageMaker Model Monitor 示例：

from sagemaker.model_monitor import DataCaptureConfig, DefaultModelMonitor
from sagemaker.model_monitor.dataset_format import DatasetFormat

# 配置数据捕获
data_capture_config = DataCaptureConfig(
    enable_capture=True,
    sampling_percentage=20,
    destination_s3_uri='s3://bucket/capture'
)

# 创建模型监控器
monitor = DefaultModelMonitor(
    role=role,
    instance_count=1,
    instance_type='ml.m5.xlarge',
    volume_size_in_gb=20,
    max_runtime_in_seconds=3600
)

# 生成基线
monitor.suggest_baseline(
    baseline_dataset='s3://bucket/train_data.csv',
    dataset_format=DatasetFormat.csv(header=True)
)

CloudWatch 自定义指标告警

import boto3

cloudwatch = boto3.client('cloudwatch', region_name='us-east-1')

# 发送自定义指标
cloudwatch.put_metric_data(
    Namespace='MLOps/ModelPerformance',
    MetricData=[
        {
            'MetricName': 'PredictionAccuracy',
            'Value': 0.94,
            'Unit': 'None',
            'Dimensions': [
                {'Name': 'ModelName', 'Value': 'fraud-detector-v2'},
                {'Name': 'Environment', 'Value': 'production'}
            ]
        }
    ]
)

9. MLOps on Cloud

GitHub Actions + AWS CodePipeline CI/CD

name: ML Pipeline CI/CD

on:
  push:
    branches: [main]

jobs:
  train-and-deploy:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Configure AWS credentials
        uses: aws-actions/configure-aws-credentials@v4
        with:
          aws-access-key-id: ${{ secrets.AWS_ACCESS_KEY_ID }}
          aws-secret-access-key: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
          aws-region: us-east-1

      - name: Run SageMaker training
        run: |
          python scripts/run_training.py \
            --instance-type ml.p3.2xlarge \
            --output-path s3://bucket/models/

      - name: Deploy to staging
        run: |
          python scripts/deploy_model.py \
            --endpoint-name staging-endpoint \
            --instance-type ml.g4dn.xlarge

MLflow + S3 模型注册表

import mlflow
import mlflow.pytorch

mlflow.set_tracking_uri('s3://bucket/mlflow')
mlflow.set_experiment('fraud-detection')

with mlflow.start_run():
    # 记录参数
    mlflow.log_params({
        'learning_rate': 0.001,
        'batch_size': 32,
        'epochs': 10
    })

    # 训练循环
    for epoch in range(10):
        train_loss = train_one_epoch(model, train_loader, optimizer)
        val_accuracy = evaluate(model, val_loader)

        mlflow.log_metrics({
            'train_loss': train_loss,
            'val_accuracy': val_accuracy
        }, step=epoch)

    # 保存模型
    mlflow.pytorch.log_model(model, 'model')
    mlflow.register_model(
        model_uri=f'runs:/{mlflow.active_run().info.run_id}/model',
        name='fraud-detector'
    )

金丝雀部署

# SageMaker 金丝雀部署
import boto3

sm = boto3.client('sagemaker')

# 把 10% 的流量切到新模型
sm.update_endpoint_weights_and_capacities(
    EndpointName='production-endpoint',
    DesiredWeightsAndCapacities=[
        {
            'VariantName': 'current-model',
            'DesiredWeight': 90,
            'DesiredInstanceCount': 4
        },
        {
            'VariantName': 'new-model',
            'DesiredWeight': 10,
            'DesiredInstanceCount': 1
        }
    ]
)

10. 测验

Q1. 在 AWS SageMaker 中启用 PyTorch DDP 分布式训练的正确配置是什么？

答案：把 distribution 参数设为 {'torch_distributed': {'enabled': True}}，并将 instance_count 设为 2 或以上。

解析：SageMaker 的 PyTorch Estimator 通过 distribution 选项支持多种分布式训练方式。torch_distributed 使用 PyTorch 原生的分布式训练框架，SageMaker 会自动完成节点间的通信配置。

Q2. Spot 实例与 On-Demand 实例的主要区别是什么？

答案：Spot 实例利用 AWS 的闲置容量，相比 On-Demand 最多可便宜 90%，但 AWS 在需要容量时会发出中断通知，并在 2 分钟内回收实例。On-Demand 随时可用，但价格是标价。

解析：把 Spot 用于 ML 训练任务时，必须实现检查点（checkpointing）。SageMaker 通过 CheckpointConfig 支持自动向 S3 做检查点，中断之后也能自动重启。

Q3. 在 BigQuery ML 的 CREATE OR REPLACE MODEL 语句中，input_label_cols 选项的含义是什么？

答案：input_label_cols 指定模型需要预测的目标列（标签）。指定的列会自动从特征中排除。

解析：BigQuery ML 直接把 SQL 查询结果用作训练数据。如果没有正确设置 input_label_cols，目标值就会被当作特征混入，造成数据泄漏（Data Leakage）。

Q4. 在 Kubernetes 中使用 KEDA 的主要目的是什么？

答案：KEDA 提供事件驱动的自动扩缩容。它根据 SQS 队列深度、Kafka 主题偏移量、HTTP 请求数等外部事件源自动调整 Pod 数量。这与默认基于 CPU/内存的 HPA（Horizontal Pod Autoscaler）不同。

解析：在 AI 推理服务中，请求队列堆积时快速扩容、空闲时缩容到 0 以节省成本，这种模式非常有效。

Q5. 把 MLflow 的模型注册表用作 S3 后端时，传给 mlflow.set_tracking_uri 的值应该是什么格式？

答案：使用 S3 URI 格式 s3://bucket-name/prefix，例如指定为 s3://my-mlflow-bucket/mlflow。

解析：MLflow 可以把 S3 用作产物存储。即使没有跟踪服务器，仅凭 S3 也能集中管理实验元数据与模型产物，这在团队级 MLOps 环境中很实用。在 EC2 或 SageMaker 环境中，可以通过 IAM 角色自动完成认证。