pie title ML 基础设施成本构成（典型比例）
"GPU 计算（训练+推理）" : 65
"存储（模型/数据/日志）" : 15
"网络与带宽" : 8
"CPU/内存（数据处理）" : 7
"平台服务与中间件" : 5
第18章 集群资源与成本治理
第18章 集群资源与成本治理
“GPU 是 AI 时代最珍贵的计算资源。浪费 GPU，就是在烧掉未来。”
18.1 为什么成本治理是 ML 平台的生存级问题
先看一组数字：
- 一张 NVIDIA H100 80GB 服务器版：约 $30,000–$40,000
- 一个 256 卡 H100 集群：约 $8M–$10M（不含网络和存储）
- 云上 8×H100 按需实例：约 $98/小时，即约 $70,000/月
如果你的团队有 50 个数据科学家，每人训练时占 4 张卡，单月 GPU 成本就可能超过 $500,000。而典型的现实是：这些 GPU 中有 30%–50% 的时间处于空闲或低效利用状态。
这意味着每个月有 $150,000–$250,000 在白白烧掉。
成本治理不是省钱——它是让每一块 GPU 都创造最大价值。
ML 成本的构成
这个比例会因场景而异。重训练场景 GPU 占比更高，重推理场景存储和带宽占比更高。但 GPU 始终是大头——这也是成本治理的首要目标。
18.2 多租户资源配额与计费模型
为什么需要多租户
ML 平台通常服务多个团队——推荐、搜索、NLP、CV 等。如果所有团队共享一个资源池而没有配额管理，必然出现以下问题：
- 吵闹邻居：一个团队提交了 64 卡的大任务，其他团队全部排队
- 成本不透明：财务问”哪个团队花了多少钱”，没人能回答
- 资源囤积：团队提前占卡但不用，“以防万一”
Kubernetes 原生配额
# namespace-quota.yaml
apiVersion: v1
kind: ResourceQuota
metadata:
name: team-research-quota
namespace: team-research
spec:
hard:
# GPU 配额（核心）
requests.nvidia.com/gpu: "16"
limits.nvidia.com/gpu: "16"
# CPU 和内存
requests.cpu: "128"
requests.memory: 512Gi
limits.cpu: "256"
limits.memory: 1Ti
# 对象数量
persistentvolumeclaims: "20"
services.loadbalancers: "2"
---
apiVersion: v1
kind: ResourceQuota
metadata:
name: team-recsys-quota
namespace: team-recsys
spec:
hard:
requests.nvidia.com/gpu: "32"
limits.nvidia.com/gpu: "32"
requests.cpu: "256"
requests.memory: 1Ti分级配额策略
单纯的硬上限会导致资源利用率低下。生产环境需要更灵活的分级配额：
# quota_manager.py
from dataclasses import dataclass, field
from typing import Dict
from datetime import datetime, timedelta
@dataclass
class TeamQuota:
team_name: str
# 保证配额：任何时候都能用到的资源
guaranteed_gpu: int
# 弹性配额：有空闲时可额外借用的上限
burst_gpu: int
# 当前使用量
current_usage: int = 0
# 历史使用记录（用于计费）
usage_history: list = field(default_factory=list)
class QuotaManager:
def __init__(self):
self.quotas: Dict[str, TeamQuota] = {}
self.total_gpu_pool = 256
def allocate(self, team: str, requested_gpu: int,
priority: str = "normal", duration_hours: float = 1.0):
"""
分级分配 GPU 资源
priority:
- guaranteed: 使用保证配额，一定成功
- burst: 使用弹性配额，需要检查全局空闲
- spot: 使用空闲资源，可被抢占
"""
quota = self.quotas[team]
if priority == "guaranteed":
if requested_gpu <= quota.guaranteed_gpu:
quota.current_usage += requested_gpu
return {"approved": True, "type": "guaranteed",
"gpu": requested_gpu}
else:
return {"approved": False,
"reason": f"超出保证配额 {quota.guaranteed_gpu}"}
elif priority == "burst":
total_used = sum(q.current_usage for q in self.quotas.values())
available = self.total_gpu_pool - total_used
if requested_gpu <= available and \
requested_gpu <= quota.burst_gpu - quota.guaranteed_gpu:
quota.current_usage += requested_gpu
return {"approved": True, "type": "burst",
"gpu": requested_gpu,
"preemptible": True} # 可被抢占
else:
return {"approved": False,
"reason": f"无足够弹性资源，当前可用 {available}"}
elif priority == "spot":
total_used = sum(q.current_usage for q in self.quotas.values())
available = self.total_gpu_pool - total_used
allocated = min(requested_gpu, available)
if allocated > 0:
quota.current_usage += allocated
return {"approved": True, "type": "spot",
"gpu": allocated,
"preemptible": True,
"max_duration_hours": duration_hours}
return {"approved": False, "reason": "无空闲资源"}
def get_usage_report(self):
"""生成使用报告"""
report = []
for team, quota in self.quotas.items():
report.append({
"team": team,
"guaranteed": quota.guaranteed_gpu,
"burst_limit": quota.burst_gpu,
"current_usage": quota.current_usage,
"utilization": quota.current_usage / quota.burst_gpu,
})
return report计费模型（Showback / Chargeback）
# billing.py - 基于 GPU 时长的计费模型
from dataclasses import dataclass
from datetime import datetime
# GPU 单价表（$/GPU·hour）
GPU_PRICING = {
"A100-80GB": 2.48, # 云上按需参考价
"H100-80GB": 4.10,
"A100-40GB": 1.86,
"V100-32GB": 1.20,
"L40S-48GB": 1.65,
# 自建摊销价（3年折旧）
"A100-80GB-self": 0.62, # ~$30k / (3*365*24*0.7 利用率)
"H100-80GB-self": 0.83,
}
@dataclass
class GPUUsageRecord:
team: str
project: str
gpu_type: str
gpu_count: int
start_time: datetime
end_time: datetime
is_self_hosted: bool = False
def compute_charge(record: GPUUsageRecord):
"""计算单次使用的费用"""
duration_hours = (record.end_time - record.start_time).total_seconds() / 3600
price_key = record.gpu_type
if record.is_self_hosted:
price_key += "-self"
unit_price = GPU_PRICING.get(price_key, 2.48)
total_cost = unit_price * record.gpu_count * duration_hours
return {
"team": record.team,
"project": record.project,
"gpu_type": record.gpu_type,
"gpu_count": record.gpu_count,
"duration_hours": round(duration_hours, 2),
"unit_price": unit_price,
"total_cost_usd": round(total_cost, 2),
"infra_type": "self-hosted" if record.is_self_hosted else "cloud",
}
# 生成月度账单
def generate_monthly_report(usage_records, month):
"""生成团队月度账单"""
from collections import defaultdict
team_costs = defaultdict(lambda: {"total": 0, "projects": defaultdict(float)})
for record in usage_records:
charge = compute_charge(record)
team_costs[charge["team"]]["total"] += charge["total_cost_usd"]
team_costs[charge["team"]]["projects"][charge["project"]] += charge["total_cost_usd"]
return dict(team_costs)
Tip
Showback vs Chargeback： - Showback：展示成本但不在组织层面实际计费。适合平台建设初期，帮助团队建立成本意识。 - Chargeback：实际将成本计入各团队预算。适合平台成熟期，通过经济杠杆驱动优化行为。 - 建议：先 Showback 3-6 个月，让团队习惯看到成本数据，再过渡到 Chargeback。
18.3 GPU 虚拟化与时间片调度
为什么需要 GPU 虚拟化
GPU 是大块头资源。一张 A100 80GB 可能价值 $20,000+，但很多推理任务只需要 5-10GB 显存。如果不做任何切分，大量显存被浪费。
GPU 共享的三种模式
graph TB
subgraph 模式1[1:1 专用]
A1[任务 A] --> G1[GPU 1]
A2[任务 B] --> G2[GPU 2]
end
subgraph 模式2[MPS 多进程服务]
B1[任务 A] --> MPS[MPS Server]
B2[任务 B] --> MPS
B3[任务 C] --> MPS
MPS --> G3[GPU 3]
end
subgraph 模式3[MIG 硬件切分]
C1[任务 A] --> MIG1[MIG 实例 1]
C2[任务 B] --> MIG2[MIG 实例 2]
C3[任务 C] --> MIG3[MIG 实例 3]
C4[任务 D] --> MIG4[MIG 实例 4]
MIG1 --> G4[GPU 4]
MIG2 --> G4
MIG3 --> G4
MIG4 --> G4
end
| 模式 | 隔离性 | 利用率 | 复杂度 | 适用场景 |
|---|---|---|---|---|
| 1:1 专用 | 最高 | 最低（~40%） | 最低 | 训练、延迟敏感推理 |
| MPS（多进程服务） | 中等 | 高（~85%） | 中等 | 多个推理服务共享 GPU |
| MIG（硬件分区） | 高 | 较高（~75%） | 较高 | 需要强隔离的推理 |
NVIDIA MIG 配置
#!/bin/bash
# mig-setup.sh - 配置 MIG 模式
# 1. 启用 MIG 模式（需要 root，GPU 空闲时操作）
sudo nvidia-smi -i 0,1,2,3 -mig 1
# 2. 选择 MIG 策略
# A100-80GB 支持以下切分方式：
# 7x 1g.10gb: 7 个实例，每个 10GB
# 3x 2g.20gb: 3 个实例，每个 20GB
# 2x 3g.40gb: 2 个实例，每个 40GB
# 1x 7g.80gb: 1 个实例，全部资源
sudo nvidia-smi mig i=0,1,2,3 gi=2,2,2,2 # 每张卡切 2 个 3g.40gb
# 3. 创建 CUDA 实例
sudo nvidia-smi mig i=0,1,2,3 ci=0,1,0,1,0,1,0,1
# 4. 验证配置
nvidia-smi mig -lgi
# 输出示例：
# +-----------------------------------------------------------------------------+
# | GPU instance ID: 0 |
# | GPU instance profile ID: 2 |
# | GPU instance profile: MIG 3g.40gb |
# | GPU instance placement: 0 |
# | Memory: 40960 MB |
# +-----------------------------------------------------------------------------+Kubernetes 中使用 MIG
# mig-deployment.yaml - 在 K8s 中使用 MIG 实例
apiVersion: apps/v1
kind: Deployment
metadata:
name: inference-service-mig
spec:
replicas: 7 # 一张 A100 切 7 个实例 → 部署 7 个推理 Pod
template:
spec:
nodeSelector:
gpu-type: A100-80GB-MIG
containers:
- name: inference
image: registry.internal/inference:v2
resources:
limits:
nvidia.com/mig-1g.10gb: 1 # 请求 1 个 MIG 1g.10gb 实例
env:
- name: MODEL_NAME
value: "bert-base-chinese"
- name: MAX_BATCH_SIZE
value: "16"时间片调度
对于不支持 MIG 的 GPU（如 V100、T4），可以使用时间片调度——多个任务轮流使用同一张 GPU：
# time_slicing.py - 基于队列的时间片调度器
import heapq
from dataclasses import dataclass, field
from datetime import datetime, timedelta
@dataclass(order=True)
class ScheduledJob:
priority: int # 数字越小，优先级越高
job_id: str
gpu_count: int
duration_minutes: int
team: str
submitted_at: datetime = field(default_factory=datetime.now)
class TimeSliceScheduler:
def __init__(self, total_gpus: int):
self.total_gpus = total_gpus
self.available_gpus = list(range(total_gpus)) # [0, 1, 2, ...]
self.running_jobs = {} # gpu_id -> (job, end_time)
self.queue = [] # 优先队列
def submit(self, job: ScheduledJob):
"""提交任务到队列"""
heapq.heappush(self.queue, job)
return self._try_schedule()
def _try_schedule(self):
"""尝试调度队列中的任务"""
scheduled = []
while self.queue and self.available_gpus:
job = heapq.heappop(self.queue)
if len(self.available_gpus) >= job.gpu_count:
# 分配 GPU
assigned_gpus = []
for _ in range(job.gpu_count):
assigned_gpus.append(self.available_gpus.pop(0))
end_time = datetime.now() + timedelta(minutes=job.duration_minutes)
for gpu in assigned_gpus:
self.running_jobs[gpu] = (job, end_time)
scheduled.append({
"job_id": job.job_id,
"gpus": assigned_gpus,
"end_time": end_time.isoformat(),
})
else:
# 资源不足，放回队列
heapq.heappush(self.queue, job)
break
return scheduled
def release_expired(self):
"""释放已完成的任务占用的 GPU"""
now = datetime.now()
released = []
for gpu_id in list(self.running_jobs.keys()):
job, end_time = self.running_jobs[gpu_id]
if now >= end_time:
self.available_gpus.append(gpu_id)
del self.running_jobs[gpu_id]
released.append({
"job_id": job.job_id,
"gpu_id": gpu_id,
"released_at": now.isoformat()
})
# 释放后重新调度
newly_scheduled = self._try_schedule()
return {"released": released, "newly_scheduled": newly_scheduled}18.4 空闲资源回收与任务合并
空闲检测
# idle_detector.py - GPU 空闲检测
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import List
@dataclass
class IdleGPUDetector:
"""检测空闲 GPU 并触发回收"""
# 判定空闲的阈值
gpu_util_threshold: float = 5.0 # GPU 利用率低于 5%
mem_util_threshold: float = 10.0 # 显存利用率低于 10%
idle_duration_minutes: int = 15 # 持续 15 分钟
def check_gpu_idle(self, gpu_metrics_history):
"""
基于 15 分钟窗口的 GPU 利用率判断是否空闲
gpu_metrics_history: [{timestamp, gpu_id, gpu_util, mem_util}, ...]
"""
cutoff = datetime.now() - timedelta(minutes=self.idle_duration_minutes)
recent = [m for m in gpu_metrics_history if m["timestamp"] >= cutoff]
idle_gpus = set()
gpu_ids = set(m["gpu_id"] for m in recent)
for gpu_id in gpu_ids:
gpu_records = [m for m in recent if m["gpu_id"] == gpu_id]
if not gpu_records:
continue
avg_gpu_util = sum(m["gpu_util"] for m in gpu_records) / len(gpu_records)
avg_mem_util = sum(m["mem_util"] for m in gpu_records) / len(gpu_records)
if avg_gpu_util < self.gpu_util_threshold and \
avg_mem_util < self.mem_util_threshold:
idle_gpus.add(gpu_id)
return list(idle_gpus)
def generate_reclaim_actions(self, idle_gpus, k8s_client):
"""生成回收动作"""
actions = []
for gpu_id in idle_gpus:
# 查找占用该 GPU 的 Pod
pods = k8s_client.get_pods_on_gpu(gpu_id)
for pod in pods:
# 检查是否是可抢占任务
if pod.annotations.get("preemptible") == "true":
actions.append({
"action": "terminate",
"pod": pod.name,
"gpu_id": gpu_id,
"reason": "idle_15min",
})
elif pod.annotations.get("preemptible") == "false":
actions.append({
"action": "notify",
"pod": pod.name,
"gpu_id": gpu_id,
"reason": "idle_15min_but_not_preemptible",
"message": f"你的任务 {pod.name} 的 GPU {gpu_id} "
f"已空闲 15 分钟，请检查或手动释放。"
})
return actions任务合并（Gang Scheduling）
多个小任务可以合并到同一组 GPU 上执行，减少资源碎片：
# job_merger.py
class JobMerger:
"""合并小的训练任务到同一组 GPU"""
def find_mergeable_jobs(self, pending_jobs):
"""
找出可以合并的任务对
合并条件：
- GPU 需求之和 <= 单节点 GPU 数量
- 内存需求之和不超限
- 优先合并同一团队的任务
"""
mergeable = []
used = set()
for i, job_a in enumerate(pending_jobs):
if i in used:
continue
for j, job_b in enumerate(pending_jobs[i+1:], start=i+1):
if j in used:
continue
if (job_a.gpu_count + job_b.gpu_count <= 8 and # 单节点 8 卡
job_a.team == job_b.team and
job_a.memory_gb + job_b.memory_gb <= 320):
mergeable.append((job_a, job_b))
used.add(i)
used.add(j)
break
return mergeable
Warning
空闲回收的陷阱： 过于激进的回收策略会导致用户对平台失去信任——他们可能在吃午饭回来后发现训练任务被杀了。始终先通知，给 15-30 分钟的宽限期，并且只回收标记为 preemptible=true 的任务。
18.5 Spot/抢占式实例利用
Spot 实例的经济价值
Spot/抢占式实例的价格通常只有按需实例的 20%–35%：
| 实例类型 | 按需价格 | Spot 价格 | 节省 |
|---|---|---|---|
p4d.24xlarge (8×A100) |
$32.77/h | ~$10.50/h | 68% |
p5.48xlarge (8×H100) |
$98.32/h | ~$30.00/h | 69% |
g5.12xlarge (4×A10G) |
$7.09/h | ~$2.20/h | 69% |
对于可以容忍中断的训练任务（如超参搜索、大规模数据预处理），Spot 实例可以大幅降低成本。
Spot 实例训练的 Checkpoint 策略
使用 Spot 实例的关键是频繁的 checkpoint——确保被抢占时能快速恢复：
# spot_training.py - 支持 Spot 中断的训练循环
import os
import torch
import signal
import time
class SpotTrainer:
def __init__(self, model, optimizer, checkpoint_dir="/checkpoints"):
self.model = model
self.optimizer = optimizer
self.checkpoint_dir = checkpoint_dir
# 监听终止信号（云厂商通常通过 SIGTERM 通知，给 2 分钟保存时间）
signal.signal(signal.SIGTERM, self._handle_termination)
self.termination_notice = False
# Checkpoint 频率
self.checkpoint_interval_steps = 100 # 每 100 步存一次
self.min_checkpoint_interval_sec = 60 # 至少间隔 1 分钟
def _handle_termination(self, signum, frame):
"""收到终止通知"""
print("⚠️ 收到 Spot 实例终止通知，正在保存 checkpoint...")
self.termination_notice = True
self._save_checkpoint(emergency=True)
def train(self, dataloader, start_step=0):
step = start_step
last_checkpoint_time = time.time()
for batch in dataloader:
# 正常训练步骤
loss = self._train_step(batch)
# 定期 checkpoint
if (step % self.checkpoint_interval_steps == 0 and
time.time() - last_checkpoint_time > self.min_checkpoint_interval_sec):
self._save_checkpoint(step=step)
last_checkpoint_time = time.time()
step += 1
# 收到终止通知后尽快退出
if self.termination_notice:
print(f"训练在第 {step} 步中断，checkpoint 已保存")
break
return step
def _save_checkpoint(self, step=None, emergency=False):
"""保存 checkpoint（优化器状态 + 模型权重 + 数据集位置）"""
step = step or 0
path = os.path.join(self.checkpoint_dir, f"ckpt_step_{step}.pt")
torch.save({
"step": step,
"model_state_dict": self.model.state_dict(),
"optimizer_state_dict": self.optimizer.state_dict(),
# 关键：记录数据集进度
"dataloader_state": dataloader.get_state(),
"timestamp": time.time(),
"emergency": emergency,
}, path)
# 上传到持久化存储（S3/OSS）
os.system(f"aws s3 cp {path} s3://ml-checkpoints/{path}")
print(f"Checkpoint 已保存: step={step}, emergency={emergency}")
def resume_from_checkpoint(self, checkpoint_path):
"""从 checkpoint 恢复"""
ckpt = torch.load(checkpoint_path)
self.model.load_state_dict(ckpt["model_state_dict"])
self.optimizer.load_state_dict(ckpt["optimizer_state_dict"])
print(f"从 step {ckpt['step']} 恢复训练")
return ckpt["step"], ckpt["dataloader_state"]Spot 实例调度器
# spot_scheduler.py
import boto3 # 或对应云 SDK
class SpotInstanceScheduler:
"""管理 Spot 实例的生命周期"""
def __init__(self, region="us-west-2"):
self.ec2 = boto3.client("ec2", region_name=region)
def request_spot_fleet(self, gpu_type="p4d.24xlarge",
target_count=4, max_price="15.0"):
"""请求 Spot 实例"""
config = {
"SpotFleetRequestConfig": {
"TargetCapacity": target_count,
"IamFleetRole": "arn:aws:iam::xxx:role/spot-fleet-role",
"LaunchSpecifications": [{
"InstanceType": gpu_type,
"SpotPrice": max_price,
"ImageId": "ami-xxx",
"SubnetId": "subnet-xxx",
# 关键：使用 EBS 挂载 checkpoint 存储
"BlockDeviceMappings": [{
"DeviceName": "/dev/xvdf",
"Ebs": {
"VolumeSize": 500,
"VolumeType": "gp3",
"Encrypted": True,
}
}],
}],
"AllocationStrategy": "capacity-optimized", # 优先选择不容易被回收的容量池
"InstanceInterruptionBehavior": "stop", # 被回收时停止而非终止
}
}
response = self.ec2.request_spot_fleet(**config)
return response["SpotFleetRequestId"]
def estimate_interruption_frequency(self, instance_type, az):
"""
估算 Spot 实例的中断频率
基于过去 30 天的历史数据
"""
# AWS 提供 Spot Instance Advisor
# 频率分为 5 档: <5%, 5-10%, 10-15%, 15-20%, >20%
frequencies = {
("p4d.24xlarge", "us-west-2a"): {"rate": "<5%", "rank": 1},
("p4d.24xlarge", "us-west-2b"): {"rate": "5-10%", "rank": 2},
("p4d.24xlarge", "us-west-2c"): {"rate": "<5%", "rank": 1},
("p5.48xlarge", "us-west-2a"): {"rate": "10-15%", "rank": 3},
}
return frequencies.get((instance_type, az),
{"rate": "unknown", "rank": 5})
Tip
Spot 实例的最佳实践： 1. 分布式训练用 capacity-optimized 策略——降低被同时回收的概率 2. 训练任务和推理任务分开——推理对延迟敏感，不应使用 Spot 3. 多可用区部署——分散到 3+ 个 AZ，降低同时中断风险 4. Checkpoint 到网络存储——不要依赖本地磁盘，Spot 实例终止后本地存储会消失 5. 设置合理的出价——不要设太低（容易被回收），也不要设太高（接近按需就失去意义）
18.6 云上与自建集群成本对比
TCO（Total Cost of Ownership）模型
# tco_calculator.py - 云上 vs 自建 TCO 对比
from dataclasses import dataclass
@dataclass
class CloudTCO:
"""云上 TCO 计算"""
hourly_cost_per_gpu: float # $/GPU·hour
gpu_count: int
utilization: float # 平均利用率 (0-1)
storage_cost_per_tb_month: float # $/TB·month
storage_tb: float
network_cost_monthly: float # $/month
months: int = 36 # 3 年 TCO
def compute(self):
# 计算成本：按实际使用时长计费
compute_cost = (
self.hourly_cost_per_gpu *
self.gpu_count *
self.utilization * 730 * # 月小时数
self.months
)
storage_cost = self.storage_cost_per_tb_month * self.storage_tb * self.months
network_cost = self.network_cost_monthly * self.months
total = compute_cost + storage_cost + network_cost
return {
"compute_cost": compute_cost,
"storage_cost": storage_cost,
"network_cost": network_cost,
"total_cost": total,
"monthly_avg": total / self.months,
"cost_per_gpu_month": total / self.months / self.gpu_count,
}
@dataclass
class OnPremTCO:
"""自建 TCO 计算"""
gpu_purchase_price: float # 单卡采购价
server_price_per_gpu: float # 每卡对应的服务器成本（CPU/内存/主板等）
gpu_count: int
network_infra_cost: float # 网络设备（交换机/线缆）
datacenter_cost_monthly: float # 机房/电力/制冷（$/月）
ops_team_cost_monthly: float # 运维团队人力（$/月）
storage_cost_per_tb: float # 存储设备采购价
storage_tb: float
depreciation_years: int = 4 # GPU 折旧年限
utilization: float = 0.7 # 平均利用率
def compute(self):
# 资本支出 (CAPEX)
gpu_capex = (self.gpu_purchase_price + self.server_price_per_gpu) * self.gpu_count
network_capex = self.network_infra_cost
storage_capex = self.storage_cost_per_tb * self.storage_tb
total_capex = gpu_capex + network_capex + storage_capex
# 运营支出 (OPEX) - 3 年
months = self.depreciation_years * 12
opex = (self.datacenter_cost_monthly + self.ops_team_cost_monthly) * months
total = total_capex + opex
return {
"gpu_capex": gpu_capex,
"network_capex": network_capex,
"storage_capex": storage_capex,
"total_capex": total_capex,
"opex": opex,
"total_cost": total,
"monthly_avg": total / months,
"cost_per_gpu_month": total / months / self.gpu_count,
"capex_ratio": total_capex / total,
}
# 示例对比：256 卡 H100 集群
cloud = CloudTCO(
hourly_cost_per_gpu=4.10, # H100 按需
gpu_count=256,
utilization=0.70, # 70% 利用率
storage_cost_per_tb_month=23, # gp3
storage_tb=500, # 500TB
network_cost_monthly=5000,
months=36,
)
onprem = OnPremTCO(
gpu_purchase_price=30000, # H100 80GB
server_price_per_gpu=15000, # 每卡对应服务器成本
gpu_count=256,
network_infra_cost=500000, # InfiniBand 网络
datacenter_cost_monthly=40000, # 机房+电力
ops_team_cost_monthly=80000, # 运维团队
storage_cost_per_tb=150, # 采购价
storage_tb=500,
depreciation_years=4,
utilization=0.70,
)
print("=== 256×H100 集群 3 年 TCO 对比 ===")
print(f"云上总成本: ${cloud.compute()['total_cost']:,.0f}")
print(f"自建总成本: ${onprem.compute()['total_cost']:,.0f}")
print(f"月均(云): ${cloud.compute()['monthly_avg']:,.0f}")
print(f"月均(自建): ${onprem.compute()['monthly_avg']:,.0f}")决策框架
graph TD
A[开始决策] --> B{GPU 规模?}
B -->|< 32 卡| C[云上按需/Spot]
B -->|32-128 卡| D{利用率 > 60%?}
B -->|> 128 卡| E{有运维团队?}
D -->|是| F[自建或云上预留]
D -->|否| G[云上 Spot + 按需混合]
E -->|是| H[自建]
E -->|否| I[云上预留 + 托管服务]
C --> J[灵活性最高，成本最高]
G --> K[性价比最优]
H --> L[长期成本最低，需要专业团队]
混合云策略
最佳实践通常是混合策略——不是非此即彼：
# hybrid-strategy.yaml - 混合云资源策略
resource_strategy:
# 基础负载（推理 + 常驻训练）：自建或预留
base_load:
type: on-premise + cloud_reserved
gpu_count: 128 # 128 卡固定
utilization_target: 75%
committed_term_months: 36
# 弹性负载（批量训练）：Spot
elastic_load:
type: cloud_spot
min_gpu_count: 0
max_gpu_count: 64 # 最多弹性扩 64 卡
spot_bid_price: "30.0" # $/instance·hour
allocation_strategy: capacity-optimized
fallback: cloud_ondemand # Spot 拿不到时降级为按需
# 突发负载（实验性任务）：按需
burst_load:
type: cloud_ondemand
max_gpu_count: 16
auto_release_minutes: 30 # 空闲 30 分钟自动释放
budget_alert_threshold: 5000 # 月度预算告警
# 存储策略
storage:
hot_data: # 活跃训练数据
type: nas
capacity_tb: 100
warm_data: # 模型 checkpoint
type: object_storage_standard
capacity_tb: 200
cold_data: # 历史数据集
type: object_storage_archive
capacity_tb: 500
retrieval_hours: 1-12
Tip
混合云的黄金法则： - 自建负责基础负载（70% 峰值需求），确保稳定供应 - Spot 负责弹性负载（20-30%），处理峰值和批量任务 - 按需负责紧急缺口（<5%），仅在 Spot 拿不到时使用 - 数据要跟算力走——跨云传数据的成本和时间可能抵消 Spot 节省的费用
18.7 小结
集群资源与成本治理是 ML 平台可持续运营的基石。核心要点：
- 多租户配额确保资源公平分配，避免吵闹邻居
- GPU 虚拟化（MIG/MPS）提升单卡利用率
- 空闲回收减少资源浪费，但要有合理的通知机制
- Spot 实例可以节省 60%+ 成本，但需要完善的 checkpoint 策略
- TCO 模型帮助做出自建 vs 云上的理性决策
记住一个核心原则：最贵的 GPU 不是单价最高的那张——而是空着不干活的那张。成本治理的第一优先级不是砍价，而是提高利用率。一张 70% 利用率的 H100 比五张 15% 利用率的 V100 更划算。
延伸阅读
- Cloud GPUs: A Comprehensive Cost Analysis (LD Wilson, 2023) — 持续更新的 GPU TCO 对比
- NVIDIA Multi-Instance GPU (MIG) 用户指南 docs.nvidia.com — MIG 官方文档
- Kubernetes Device Plugins kubernetes.io — GPU 共享的 K8s 方案
- AWS Spot Instance Advisor aws.amazon.com/ec2/spot — Spot 中断频率和最佳实践
- FinOps for Machine Learning (FinOps Foundation, 2024) — ML 场景下的云财务管理框架
- MLPerf Inferring Benchmark mlcommons.org — GPU 推理性能基准，辅助采购决策