freeCodeCamp 1小时前 · 2026-09-09 17:44:50 · 1 阅读
AWS 云成本监控、告警与优化：开发者实战指南
工程团队里有个每月都上演的经典对话：有人把 AWS 账单截图丢到群里，数字比上个月又高了，大家点点头，一致认为该降一降，然后就没有然后了——下个月照常循环。
这篇文章的目标就是终结这个循环，用一套系统化的、按服务逐项拆解的方法取而代之：搞清楚你花了多少钱、为什么花、在问题变成账单之前就拦住它、砍成本不靠拍脑袋。
全文按"可反复查阅"的思路组织，每一部分都自成体系：当前问题在 RDS，可以直接跳到 RDS 那一节；从零搭建 FinOps 实践，则从头读到尾即可。所有命令都可以直接跑，所有脚本都可以直接部署。
目录
你将学到什么
如何用 Athena 查询 Cost and Usage Report——这是所有严肃成本分析的基石
每个工程团队都该搭的五张仪表盘，附带今天就能跑通的查询语句
三层告警策略，既能抓住成本尖刺，又不会让团队陷入告警疲劳
EC2、S3、Lambda、RDS、DynamoDB、数据传输的逐项优化手册
一份可落地的 30 天冲刺计划，第一个月就能看到可量化的节省
下面从零开始，一步步搭起来。
前置条件
在开始之前，请确保你具备以下条件：
知识层面：
熟悉 AWS 常用服务：EC2、S3、RDS、Lambda、VPC
能阅读 Python 和 SQL 代码
了解 IAM 策略和角色的基本机制
访问权限：
一个有账单访问权限的 AWS 账号。你使用的 IAM 用户或角色需要
ce:GetCostAndUsage、ec2:Describe*、rds:Describe*、s3:GetBucketLifecycleConfiguration这几项权限。已配置 AWS CLI v2
Athena 访问权限（Part 1 中的 CUR 查询会用到）
准备工作：
- 如果尚未开启 Cost Explorer，先启用它。它免费，且本指南中大部分命令都依赖它：
aws ce enable-cost-explorer --region us-east-1
- 确保 Cost and Usage Report（CUR）已配置并导出到 S3 桶。如果还没有，可以按 AWS CUR 配置指南 操作。配置完成后等 24 小时，报告才会生成第一个数据文件。
Part 1：监控——搞清楚每笔钱花在哪了
1.1 Cost and Usage Report——你的数据基准
Cost Explorer 展示的是服务级别的汇总数据，看趋势够用，但做根因分析不够。当你需要定位到底是哪个资源导致了每月 12,000 美元的某笔费用时，就得通过 Athena 查询 CUR。
在 CUR 数据上创建 Athena 表：
-- CUR 开始生成数据后，在 Athena 里运行一次
-- 把 'your-cur-bucket' 和 'your-prefix' 替换为你的实际值
CREATE EXTERNAL TABLE IF NOT EXISTS cur_database.billing (
bill_billing_period_start_date STRING,
bill_payer_account_id STRING,
line_item_usage_start_date STRING,
line_item_resource_id STRING,
line_item_usage_type STRING,
line_item_usage_amount DOUBLE,
line_item_unblended_cost DOUBLE,
product_servicecode STRING,
product_instance_type STRING,
product_region STRING,
resource_tags_user_environment STRING,
resource_tags_user_team STRING,
resource_tags_user_service STRING,
resource_tags_user_owner STRING
)
PARTITIONED BY (year STRING, month STRING)
ROW FORMAT DELIMITED
FIELDS TERMINATED BY ','
LOCATION 's3://your-cur-bucket/your-prefix/'
TBLPROPERTIES ('skip.header.line.count'='1');
MSCK REPAIR TABLE cur_database.billing;
以下是第一天就该跑的三条查询：
-- 查询 1：本月成本最高的前 20 个资源
-- 先跑这条，它能告诉你该重点关注哪里。
SELECT
line_item_resource_id,
product_servicecode,
resource_tags_user_team AS team,
resource_tags_user_service AS service,
SUM(line_item_unblended_cost) AS total_cost_usd
FROM cur_database.billing
WHERE line_item_usage_start_date >= DATE_FORMAT(
DATE_TRUNC('month', CURRENT_DATE), '%Y-%m-%d'
)
AND line_item_unblended_cost > 0
GROUP BY 1, 2, 3, 4
ORDER BY total_cost_usd DESC
LIMIT 20;
-- 查询 2：按服务查看周环比成本增长
-- 识别成本增长最快的服务，这些需要重点排查
WITH weekly AS (
SELECT
DATE_TRUNC('week', CAST(line_item_usage_start_date AS DATE)) AS week,
product_servicecode,
SUM(line_item_unblended_cost) AS cost
FROM cur_database.billing
WHERE line_item_usage_start_date >=
DATE_FORMAT(DATE_ADD('day', -42, CURRENT_DATE), '%Y-%m-%d')
GROUP BY 1, 2
)
SELECT
curr.product_servicecode AS service,
ROUND(prev.cost, 2) AS prev_week_cost,
ROUND(curr.cost, 2) AS curr_week_cost,
ROUND(
100.0 * (curr.cost - prev.cost) / NULLIF(prev.cost, 0),
1
) AS pct_change
FROM weekly curr
JOIN weekly prev
ON curr.product_servicecode = prev.product_servicecode
AND curr.week = DATE_ADD('week', 1, prev.week)
WHERE curr.week = DATE_TRUNC('week', CURRENT_DATE)
AND ABS((curr.cost - prev.cost) / NULLIF(prev.cost, 0)) > 0.20
ORDER BY pct_change DESC;
-- 查询 3：有成本但无使用的资源（可关停候选）
-- 查找过去 7 天内产生了费用但用量为零的资源
-- 这是资源处于空闲或孤立状态的强烈信号
SELECT
line_item_resource_id,
product_servicecode,
resource_tags_user_owner AS owner,
resource_tags_user_team AS team,
SUM(line_item_unblended_cost) AS cost_past_7_days
FROM cur_database.billing
WHERE line_item_usage_start_date >=
DATE_FORMAT(DATE_ADD('day', -7, CURRENT_DATE), '%Y-%m-%d')
AND line_item_usage_amount = 0
AND line_item_unblended_cost > 5
GROUP BY 1, 2, 3, 4
ORDER BY cost_past_7_days DESC
LIMIT 30;
1.2 五个必备的成本看板
这五个视图覆盖了大多数工程团队的监控需求，每个都由可直接执行的查询构建，无需依赖第三方工具。
看板 1：高管摘要（用于每周管理层汇报）
# executive_summary.py
import boto3
from datetime import datetime, timedelta
ce = boto3.client('ce')
def weekly_summary():
today = datetime.now()
start_mtd = today.replace(day=1).strftime('%Y-%m-%d')
today_str = today.strftime('%Y-%m-%d')
# 当月累计支出
mtd = ce.get_cost_and_usage(
TimePeriod={'Start': start_mtd, 'End': today_str},
Granularity='MONTHLY',
Metrics=['UnblendedCost']
)
mtd_spend = float(
mtd['ResultsByTime'][0]['Total']['UnblendedCost']['Amount']
)
# 月末费用预测
forecast = ce.get_cost_forecast(
TimePeriod={
'Start': today_str,
'End': (today.replace(day=28) + timedelta(days=4)).replace(day=1).strftime('%Y-%m-%d'),
},
Metric='UNBLENDED_COST',
Granularity='MONTHLY'
)
eom_forecast = float(forecast['Total']['Amount']) + mtd_spend
# 前五大服务
by_service = ce.get_cost_and_usage(
TimePeriod={'Start': start_mtd, 'End': today_str},
Granularity='MONTHLY',
Metrics=['UnblendedCost'],
GroupBy=[{'Type': 'DIMENSION', 'Key': 'SERVICE'}]
)
services = sorted(
[
(g['Keys'][0], float(g['Metrics']['UnblendedCost']['Amount']))
for g in by_service['ResultsByTime'][0]['Groups']
],
key=lambda x: x[1],
reverse=True
)[:5]
print(f"\n{'─'*48}")
print(f" AWS Cost Summary — {today.strftime('%B %Y')}")
print(f"{'─'*48}")
print(f" Month-to-date: ${mtd_spend:>12,.2f}")
print(f" End-of-month est: ${eom_forecast:>12,.2f}")
print(f"\n Top 5 Services:")
for name, cost in services:
short = name.replace('Amazon ', '').replace('AWS ', '')
print(f" {short:<32} ${cost:>9,.2f}")
print(f"{'─'*48}\n")
weekly_summary()
仪表盘 2：团队费用明细（面向工程负责人）
# team_breakdown.py
import boto3
from datetime import datetime
ce = boto3.client('ce')
def team_breakdown():
start = datetime.now().replace(day=1).strftime('%Y-%m-%d')
end = datetime.now().strftime('%Y-%m-%d')
response = ce.get_cost_and_usage(
TimePeriod={'Start': start, 'End': end},
Granularity='MONTHLY',
Metrics=['UnblendedCost'],
GroupBy=[
{'Type': 'TAG', 'Key': 'Team'},
{'Type': 'DIMENSION', 'Key': 'SERVICE'},
]
)
by_team = {}
for group in response['ResultsByTime'][0].get('Groups', []):
team_raw = group['Keys'][0]
team = team_raw.replace('Team$', '') if team_raw else 'untagged'
service = group['Keys'][1]
cost = float(group['Metrics']['UnblendedCost']['Amount'])
if team not in by_team:
by_team[team] = {'total': 0.0, 'by_service': {}}
by_team[team]['total'] += cost
by_team[team]['by_service'][service] = (
by_team[team]['by_service'].get(service, 0.0) + cost
)
total_bill = sum(d['total'] for d in by_team.values())
print(f"\n{'─'*58}")
print(f" Team Cost Breakdown — MTD {datetime.now().strftime('%Y-%m-%d')}")
print(f" Total: ${total_bill:,.2f}")
print(f"{'─'*58}")
for team, data in sorted(by_team.items(), key=lambda x: x[1]['total'], reverse=True):
pct = (data['total'] / total_bill * 100) if total_bill else 0
print(f"\n {team:<20} ${data['total']:≥10,.2f} ({pct:.1f}%)")
top3 = sorted(data['by_service'].items(), key=lambda x: x[1], reverse=True)[:3]
for svc, cost in top3:
short = svc.replace('Amazon ', '').replace('AWS ', '')
print(f" └─ {short:<30} ${cost:>8,.2f}")
print()
team_breakdown()
看板 3：浪费检测（用于每周清理回顾）：
# waste_detector.py
import boto3
from datetime import datetime, timezone, timedelta
ec2 = boto3.client('ec2')
elbv2 = boto3.client('elbv2')
cw = boto3.client('cloudwatch')
def detect_waste():
report = {'items': [], 'total_monthly_waste': 0.0}
# 未挂载的 EBS 卷
for vol in ec2.describe_volumes(
Filters=[{'Name': 'status', 'Values': ['available']}]
)['Volumes']:
age = (datetime.now(timezone.utc) - vol['CreateTime']).days
cost = round(vol['Size'] * 0.08, 2)
tags = {t['Key']: t['Value'] for t in vol.get('Tags', [])}
report['items'].append({
'type': '未挂载的 EBS 卷',
'id': vol['VolumeId'],
'detail': f"{vol['Size']}GB — 已存在 {age} 天",
'owner': tags.get('Owner', '—'),
'monthly_cost': cost,
})
report['total_monthly_waste'] += cost
# 未关联的 Elastic IP
for addr in ec2.describe_addresses()['Addresses']:
if 'AssociationId' not in addr:
report['items'].append({
'type': '未关联的 Elastic IP',
'id': addr.get('AllocationId', ''),
'detail': addr['PublicIp'],
'owner': '—',
'monthly_cost': 3.60,
})
report['total_monthly_waste'] += 3.60
# 闲置负载均衡器（7 天内请求数不足 100）
for lb in elbv2.describe_load_balancers()['LoadBalancers']:
metrics = cw.get_metric_statistics(
Namespace='AWS/ApplicationELB',
MetricName='RequestCount',
Dimensions=[{'Name': 'LoadBalancer',
'Value': lb['LoadBalancerArn'].split(':loadbalancer/')[-1]}],
StartTime=datetime.now() - timedelta(days=7),
EndTime=datetime.now(),
Period=604800,
Statistics=['Sum']
)['Datapoints']
total_requests = metrics[0]['Sum'] if metrics else 0
if total_requests < 100:
report['items'].append({
'type': '闲置负载均衡器',
'id': lb['LoadBalancerName'],
'detail': f"7 天内共 {int(total_requests)} 次请求",
'owner': '—',
'monthly_cost': 22.0,
})
report['total_monthly_waste'] += 22.0
print(f"\n 资源浪费检测报告 — {datetime.now().strftime('%Y-%m-%d')}")
print(f" 预估月度浪费：${report['total_monthly_waste']:.2f}\n")
for item in sorted(report['items'], key=lambda x: x['monthly_cost'], reverse=True)[:20]:
print(f" [{item['type']}]")
print(f" ID：{item['id']}")
print(f" 详情：{item['detail']}")
print(f" 负责人：{item['owner']}")
print(f" 费用：${item['monthly_cost']:.2f}/月\n")
return report
detect_waste()
1.3 标签策略——所有归因的基石
任何成本归因模型都依赖标签。跳过打标签的团队搭出来的仪表盘只能看到总额，无法解释原因。关键不在于提醒团队"记得打标签"，而在于把标签变成基础设施代码的硬约束，而不是 Confluence 里一条流程规范。
必选标签集：
# terraform/variables.tf
variable "mandatory_tags" {
description = "Tags applied to every resource in this account"
type = map(string)
validation {
condition = alltrue([
contains(keys(var.mandatory_tags), "Environment"),
contains(keys(var.mandatory_tags), "Team"),
contains(keys(var.mandatory_tags), "Owner"),
contains(keys(var.mandatory_tags), "Service"),
])
error_message = "mandatory_tags must include Environment, Team, Owner, and Service."
}
}
locals {
common_tags = merge(var.mandatory_tags, {
ManagedBy = "terraform"
LastModified = timestamp()
})
}
resource "aws_instance" "api_server" {
ami = data.aws_ami.amazon_linux_2023.id
instance_type = "t3.medium"
tags = merge(local.common_tags, {Name = "api-server-${var.environment}"})
}
每周排查并报告未打标签的资源：
#!/usr/bin/env bash
# find_untagged.sh
echo "Untagged EC2 instances (missing Team tag):"
aws ec2 describe-instances \
--filters "Name=instance-state-name,Values=running" \
--query "Reservations[].Instances[?!not_null(Tags[?Key=='Team'].Value|[0])].[InstanceId,InstanceType,LaunchTime]" \
--output table
echo "Untagged RDS instances:"
aws rds describe-db-instances \
--query "DBInstances[?!not_null(TagList[?Key=='Team'].Value|[0])].DBInstanceIdentifier" \
--output table
第二部分：告警——在尖峰变成账单之前抓住它
没有主动告警时，典型的发现时间线是这样的：5 号发生费用尖峰，20 号月度账单到账，22 号才有人注意到，23 号开始排查——那两周的浪费已经收不回来了。而有了主动告警，发现时间可以压缩到数小时内。
2.1 三级告警结构
告警疲劳和没有告警一样致命。三级模型的核心思路：把不同严重级别的路由到不同渠道，对应不同的响应预期，从而保持每条告警的信号强度。
# alert_router.py
import boto3
import json
import urllib.request
from enum import Enum
SLACK_INFO_WEBHOOK = 'https://hooks.slack.com/services/INFO/WEBHOOK'
SLACK_ALERT_WEBHOOK = 'https://hooks.slack.com/services/ALERT/WEBHOOK'
SNS_CRITICAL_TOPIC = 'arn:aws:sns:us-east-1:YOUR_ACCOUNT:cost-critical'
class AlertTier(Enum):
INFO = 1
WARNING = 2
CRITICAL = 3
def route_alert(tier: AlertTier, subject: str, message: str):
"""按严重级别将告警发送到对应的渠道。"""
icons = {AlertTier.INFO: ':information_source:',
AlertTier.WARNING: ':warning:', AlertTier.CRITICAL: ':rotating_light:'}
payload = {'text': f"{icons[tier]} *{subject}*\n{message}"}
if tier == AlertTier.INFO:
_post_slack(SLACK_INFO_WEBHOOK, payload)
elif tier == AlertTier.WARNING:
_post_slack(SLACK_ALERT_WEBHOOK, payload)
_send_sns(SNS_CRITICAL_TOPIC, subject, f"WARNING: {message}")
elif tier == AlertTier.CRITICAL:
_post_slack(SLACK_ALERT_WEBHOOK, payload)
_send_sns(SNS_CRITICAL_TOPIC, subject, f"CRITICAL: {message}")
def _post_slack(webhook: str, payload: dict):
req = urllib.request.Request(
webhook,
data=json.dumps(payload).encode(),
headers={'Content-Type': 'application/json'}
)
urllib.request.urlopen(req)
def _send_sns(topic_arn: str, subject: str, message: str):
sns = boto3.client('sns')
sns.publish(TopicArn=topic_arn, Subject=subject[:100], Message=message)
三个级别的分工如下：Level 1 INFO 发送到 Slack 信息频道，用于每日成本汇总、每周趋势报告和标签合规更新，无需处理。
Level 2 WARNING 发送到 Slack 告警频道，同时发送邮件，触发条件包括预算使用率超过 75%、成本周环比上涨 25%、Savings Plans 即将到期，需在 24 小时内确认。
Level 3 CRITICAL 触发 PagerDuty 和短信，条件包括预算使用率超过 90%、24 小时内成本翻倍、检测到挖矿行为、预计超支超过计划的 120%，需在 1 小时内排查。
2.2 实时预算监控
AWS Budgets 默认每天只发一次告警。按天为窗口意味着，早上 8 点开始的成本飙升，要等到第二天的告警才会被发现。下面的 Lambda 每小时运行一次，同时检查预算的绝对使用率和逐小时的变化速率。
# budget_monitor.py
# 由 EventBridge 每小时触发的 Lambda 函数
import boto3
from datetime import datetime, timedelta
from alert_router import route_alert, AlertTier
ce = boto3.client('ce')
budgets = boto3.client('budgets', region_name='us-east-1')
ACCOUNT_ID = boto3.client('sts').get_caller_identity()['Account']
BUDGET_NAME = 'monthly-infrastructure'
def get_mtd_spend() -> float:
start = datetime.now().replace(day=1).strftime('%Y-%m-%d')
end = datetime.now().strftime('%Y-%m-%d')
r = ce.get_cost_and_usage(
TimePeriod={'Start': start, 'End': end},
Granularity='MONTHLY',
Metrics=['UnblendedCost']
)
return float(r['ResultsByTime'][0]['Total']['UnblendedCost']['Amount'])
def get_budget_limit() -> float:
r = budgets.describe_budget(AccountId=ACCOUNT_ID, BudgetName=BUDGET_NAME)
return float(r['Budget']['BudgetLimit']['Amount'])
def get_hourly_costs(hours: int = 4) -> list:
"""返回最近 N 小时每小时的总费用。"""
end = datetime.now()
start = end - timedelta(hours=hours)
r = ce.get_cost_and_usage(
TimePeriod={'Start': start.strftime('%Y-%m-%d'), 'End': end.strftime('%Y-%m-%d')},
Granularity='HOURLY',
Metrics=['UnblendedCost']
)
return [
float(period['Total']['UnblendedCost']['Amount'])
for period in r['ResultsByTime']
]
def lambda_handler(event, context):
mtd_spend = get_mtd_spend()
budget_limit = get_budget_limit()
utilisation = mtd_spend / budget_limit * 100
days_elapsed = datetime.now().day
projected_eom = (mtd_spend / days_elapsed) * 30
projected_pct = projected_eom / budget_limit * 100
if utilisation >= 90:
route_alert(
AlertTier.CRITICAL,
f'Budget at {utilisation:.0f}%',
f'MTD spend ${mtd_spend:,.2f} is {utilisation:.0f}% of ${budget_limit:,.0f} budget. '
f'Projected EOM: ${projected_eom:,.2f}.'
)
elif utilisation >= 75:
route_alert(
AlertTier.WARNING,
f'Budget at {utilisation:.0f}%',
f'MTD spend ${mtd_spend:,.2f} is {utilisation:.0f}% of ${budget_limit:,.0f} budget. '
f'Projected EOM: ${projected_eom:,.2f}.'
)
# 检查小时级费用尖峰
hourly = get_hourly_costs(hours=4)
if len(hourly) >= 2:
last_hour = hourly[-1]
prev_avg = sum(hourly[:-1]) / len(hourly[:-1])
if prev_avg > 0.10 and last_hour > prev_avg * 1.5:
route_alert(
AlertTier.WARNING,
'Hourly cost spike detected',
f'Last hour: ${last_hour:.2f} vs prior 3-hour avg ${prev_avg:.2f} '
f'(+{(last_hour/prev_avg - 1)*100:.0f}%)'
)
return {
'mtd_spend': round(mtd_spend, 2),
'utilisation_pct': round(utilisation, 1),
'projected_eom': round(projected_eom, 2),
}
第三部分：按服务优化
3.1 EC2 — 七个优先调控手段
EC2 是大多数 AWS 账户中占比最大的开销项，也是优化空间最大的服务。请按顺序逐一执行以下调控手段，因为每一步都会降低下一步的基准成本。
调控 1：找出真正空闲的实例（连续 14 天 CPU 低于 1%）。
# ec2_idle_finder.py
import boto3
from datetime import datetime, timedelta
ec2 = boto3.client('ec2')
cw = boto3.client('cloudwatch')
def find_idle_instances(avg_cpu_threshold: float = 1.0, days: int = 14):
instances = [
inst
for r in ec2.describe_instances(
Filters=[{'Name': 'instance-state-name', 'Values': ['running']}]
)['Reservations']
for inst in r['Instances']
]
idle = []
for inst in instances:
iid = inst['InstanceId']
stats = cw.get_metric_statistics(
Namespace='AWS/EC2',
MetricName='CPUUtilization',
Dimensions=[{'Name': 'InstanceId', 'Value': iid}],
StartTime=datetime.utcnow() - timedelta(days=days),
EndTime=datetime.utcnow(),
Period=days * 86400,
Statistics=['Average']
)['Datapoints']
avg_cpu = stats[0]['Average'] if stats else 0.0
if avg_cpu < avg_cpu_threshold:
tags = {t['Key']: t['Value'] for t in inst.get('Tags', [])}
idle.append({
'instance_id': iid,
'instance_type': inst['InstanceType'],
'avg_cpu': round(avg_cpu, 2),
'environment': tags.get('Environment', '—'),
'owner': tags.get('Owner', '—'),
})
return sorted(idle, key=lambda x: x['avg_cpu'])
for inst in find_idle_instances():
print(f" {inst['instance_id']} {inst['instance_type']} "
f"{inst['avg_cpu']}% CPU env:{inst['environment']} owner:{inst['owner']}")
调控 2：缩容过度配置的实例（CPU 持续低于 20%）。
使用同一脚本，将 avg_cpu_threshold 设为 20.0。这些是缩容候选，而非停机候选。
调控 3：利用 EventBridge 规则，为标记了 AutoShutdown=true 的实例排定开发和预发环境的定时停机。
调控 4：完成调控 1–3 后再购买 Savings Plans。
调控 5：迁移到 Graviton（成本降低 20%，绝大多数负载性能不变）。
手段 6：对容错型批处理和开发类工作负载使用 Spot 实例。
手段 7：把容器化工作负载迁移到 EKS，配合 Karpenter 实现自动装箱。
Spot 节省估算：
# spot_price_analyser.py
import boto3
ec2 = boto3.client('ec2')
def spot_savings_estimate(instance_type: str) -> dict:
spot_history = ec2.describe_spot_price_history(
InstanceTypes=[instance_type],
ProductDescriptions=['Linux/UNIX'],
MaxResults=1
)['SpotPriceHistory']
spot_price = float(spot_history[0]['SpotPrice']) if spot_history else 0
on_demand_approx = {
't3.medium': 0.0416, 'm5.large': 0.096,
'c5.xlarge': 0.17, 'r5.2xlarge': 0.504,
}
od_price = on_demand_approx.get(instance_type, 0)
savings_pct = ((od_price - spot_price) / od_price * 100) if od_price else 0
return {
'instance_type': instance_type,
'spot_price': round(spot_price, 4),
'on_demand': od_price,
'savings_pct': round(savings_pct, 1),
'monthly_spot': round(spot_price * 730, 2),
'monthly_od': round(od_price * 730, 2),
}
for itype in ['t3.medium', 'm5.large', 'c5.xlarge']:
r = spot_savings_estimate(itype)
print(f" {r['instance_type']:<15} Spot: ${r['spot_price']}/hr "
f"OD: ${r['on_demand']}/hr Savings: {r['savings_pct']}%")
3.2 S3——生命周期策略与存储类别选择
S3 优化包括两方面：通过生命周期策略把低频访问的数据迁移到更便宜的存储类别，以及消除浪费模式，比如未完成的分段上传（incomplete multipart uploads）。
# s3_lifecycle_applier.py
import boto3
s3 = boto3.client('s3')
LOG_POLICY = {
'Rules': [{
'ID': 'standard-tiering',
'Status': 'Enabled',
'Filter': {'Prefix': ''},
'Transitions': [
{'Days': 30, 'StorageClass': 'STANDARD_IA'},
{'Days': 90, 'StorageClass': 'GLACIER_IR'},
{'Days': 365, 'StorageClass': 'DEEP_ARCHIVE'},
],
'Expiration': {'Days': 2555},
'AbortIncompleteMultipartUpload': {'DaysAfterInitiation': 7},
}]
}
TEMP_POLICY = {
'Rules': [{
'ID': 'temp-data-retention',
'Status': 'Enabled',
'Filter': {'Prefix': ''},
'Expiration': {'Days': 30},
'AbortIncompleteMultipartUpload': {'DaysAfterInitiation': 1},
}]
}
for bucket in s3.list_buckets()['Buckets']:
name = bucket['Name']
try:
s3.get_bucket_lifecycle_configuration(Bucket=name)
print(f" {name} — policy already exists, skipping")
except s3.exceptions.ClientError:
policy = TEMP_POLICY if any(k in name for k in ['temp', 'build', 'cache']) else LOG_POLICY
s3.put_bucket_lifecycle_configuration(
Bucket=name, LifecycleConfiguration=policy
)
print(f" {name} — applied {'TEMP' if policy is TEMP_POLICY else 'LOG'} policy")
3.3 RDS — 五级优化方案
| 级别 | 操作 | 典型节省 | 风险 |
|---|---|---|---|
| 1 | 删除未使用的只读副本 | 副本费用的 30%–50% | 低 |
| 2 | 将备份保留期缩短至合规最低要求 | 存储费用的 20%–30% | 低 |
| 3 | 按需降配实例规格（CPU 持续低于 20%） | 计算费用的 20%–40% | 中 |
| 4 | 为生产环境购买 Reserved Instances | 计算费用的 30%–60% | 低 |
| 5 | 将负载波动较大的数据库迁移至 Aurora Serverless v2 | 总体费用 40%–70% | 工作量大 |
查找过度配置的 RDS 实例：
# rds_rightsizer.py
import boto3
from datetime import datetime, timedelta
rds = boto3.client('rds')
cw = boto3.client('cloudwatch')
def find_oversized_rds():
instances = rds.describe_db_instances()['DBInstances']
candidates = []
for inst in instances:
iid = inst['DBInstanceIdentifier']
iclass = inst['DBInstanceClass']
stats = cw.get_metric_statistics(
Namespace='AWS/RDS',
MetricName='CPUUtilization',
Dimensions=[{'Name': 'DBInstanceIdentifier', 'Value': iid}],
StartTime=datetime.utcnow() - timedelta(days=14),
EndTime=datetime.utcnow(),
Period=1209600,
Statistics=['Average', 'Maximum']
)['Datapoints']
if not stats:
continue
avg_cpu = stats[0]['Average']
max_cpu = stats[0]['Maximum']
if avg_cpu < 20 and max_cpu < 50:
candidates.append({
'id': iid,
'class': iclass,
'avg_cpu': round(avg_cpu, 1),
'max_cpu': round(max_cpu, 1),
'engine': inst['Engine'],
})
return candidates
for c in find_oversized_rds():
print(f" {c['id']} {c['class']} avg:{c['avg_cpu']}% max:{c['max_cpu']}% engine:{c['engine']}")
3.4 DynamoDB：On-Demand 与 Provisioned 的选型
On-Demand 模式省心，但对于负载可预测的场景，成本可能是 Provisioned 的 3–5 倍。Provisioned 搭配自动扩缩容能覆盖绝大多数情况，成本显著更低。
# dynamodb_mode_advisor.py
import boto3
from datetime import datetime, timedelta
dynamodb = boto3.client('dynamodb')
cw = boto3.client('cloudwatch')
def analyse_table_billing(table_name: str) -> dict:
table = dynamodb.describe_table(TableName=table_name)['Table']
mode = table.get('BillingModeSummary', {}).get('BillingMode', 'PROVISIONED')
stats = {}
for metric in ['ConsumedReadCapacityUnits', 'ConsumedWriteCapacityUnits']:
data = cw.get_metric_statistics(
Namespace='AWS/DynamoDB',
MetricName=metric,
Dimensions=[{'Name': 'TableName', 'Value': table_name}],
StartTime=datetime.utcnow() - timedelta(days=30),
EndTime=datetime.utcnow(),
Period=86400,
Statistics=['Average', 'Maximum']
)['Datapoints']
if data:
avg = sum(d['Average'] for d in data) / len(data)
peak = max(d['Maximum'] for d in data)
stats[metric] = {'avg': round(avg, 1), 'peak': round(peak, 1)}
if mode == 'PAY_PER_REQUEST':
avg_rcu = stats.get('ConsumedReadCapacityUnits', {}).get('avg', 0)
avg_wcu = stats.get('ConsumedWriteCapacityUnits', {}).get('avg', 0)
if avg_rcu > 2000 or avg_wcu > 500:
return {
'table': table_name,
'current_mode': 'PAY_PER_REQUEST',
'recommendation': 'Switch to PROVISIONED with auto-scaling',
'reason': f'Avg {avg_rcu:.0f} RCU/s and {avg_wcu:.0f} WCU/s — predictable pattern',
}
return {'table': table_name, 'current_mode': mode, 'recommendation': 'No change needed'}
for table in dynamodb.list_tables()['TableNames']:
r = analyse_table_billing(table)
if r['recommendation'] != 'No change needed':
print(f" {r['table']}: {r['recommendation']}")
3.5 数据传输 — 三种主要浪费模式
数据传输费用往往是 AWS 账单里最让人看不懂的一项。以下是三种主要模式及其解决办法：
跨可用区流量（最常见，也最容易解决）：不同 AZ 之间的服务往返都要收取 $0.01/GB 的费用。解决办法是在 Kubernetes Services 上启用拓扑感知路由，或者让应用层和数据库层部署在同一个 AZ。
NAT Gateway 对内部 AWS 流量收费：经过 NAT Gateway 路由的 S3、ECR、DynamoDB、SQS 流量按 $0.045/GB 计费。解决办法：用 VPC endpoints 彻底消除这项费用。
合规不需要的跨区域复制：每季度对照实际合规要求审查 S3 复制规则，清理多余配置。
# 查找开启了复制的 S3 存储桶
for bucket in $(aws s3api list-buckets --query 'Buckets[*].Name' --output text); do
result=$(aws s3api get-bucket-replication --bucket "$bucket" 2>&1)
if ! echo "$result" | grep -q "ReplicationConfigurationNotFoundError"; then
echo " $bucket — 复制已开启，请确认合规需求"
fi
done
第四部分：30 天优化冲刺
这个冲刺在第一个月就能带来可量化的节省。设计目标是：一名工程师每周投入 2–4 小时的 FinOps 专项时间。
第一周——建立可见性：开启 CUR，建好 Athena 表，跑完三组首日查询并截图存档作为基线，给所有运行中的 EC2 和 RDS 实例补齐标签，找出前三大成本驱动因素并逐一记录假设。
第二周——快速见效：部署孤儿资源上报 Lambda，给最大的三个 S3 存储桶配置生命周期策略，运行空闲实例检测器，将平均 CPU 低于 1% 且负责人无异议的实例停掉，并为 S3、ECR、DynamoDB 添加 VPC endpoints。
第三周——规格优化：运行 EC2 规格分析器，将置信度最高的三个候选缩容（非生产环境优先），跑 RDS 规格优化脚本，找出流量极低的只读副本并下线。
第四周——告警与自动化：部署每小时预算监控 Lambda，配置三级告警路由，建好每周浪费报告，在基础设施仓库中加入 Infracost GitHub Action，并安排每月 30 分钟的 FinOps 复盘会。
30 天预期效果：月 AWS 支出降低 15–25%，每项变更均有文档记录，并形成可持续的例行流程，防止同类浪费再次累积。
最佳实践速览
✅ 应该做：先搭好 CUR + Athena，再做其他监控。Cost Explorer 只是起点，CUR 才是数据源头。
✅ 该做的：在 Terraform 或 CloudFormation 中强制使用标签。靠流程约定打标签会慢慢失效，基础设施层面的强制标签才一劳永逸。
✅ 该做的：每周跑一次闲置实例查找器和浪费报告器。浪费是持续累积的，每周出一份报告才能把"欠账"控制住。
✅ 该做的：采用三层告警模型。把所有告警塞进同一个通道只会让人疲劳，最终被一键静音。
✅ 该做的：按顺序使用 EC2 各项优化手段。先做 Right-sizing 再签 Savings Plans，避免把浪费以"优惠价"的形式锁定下来。
✅ 该做的：每季度核对一次 DynamoDB 计费模式与实际使用模式的匹配度。
❌ 别做：未经调查就删除未打标签的资源。没打标签不等于没用，只代表没人认领。
❌ 别做：没审计访问模式就上激进的 S3 生命周期策略。如果数据访问频率高于预期，Glacier 取回费用可能比 Standard 存储成本还贵。
❌ 别做：浪费报告器 Lambda 首次部署就开启自动删除。先以纯报告模式跑两周，确认输出无误后再加入删除逻辑。
参考资料
AWS Cost and Usage Report 数据字典：本指南中所有 CUR Athena 查询的字段参考
AWS Cost Explorer API 参考：Python boto3 成本查询接口的完整文档
AWS Compute Optimizer：基于 ML 的 Right-sizing 推荐，可用来交叉验证手动分析脚本的结果
Amazon DynamoDB 定价：3.4 节中预置容量与按需计费成本对比的权威参考
AWS Instance Scheduler：AWS 官方的基于标签的 EC2 和 RDS 定时调度方案
FinOps Foundation Framework：实践者框架，定义了本指南所实现的 Inform、Optimise、Operate 循环
配套代码仓库：本指南用到的所有脚本、Lambda 函数和 Terraform 模块
原始来源： freeCodeCamp