如何在AWS Lambda中用Python将Cost Explorer数据生成CSV文件

我来一步步帮你解决这两个问题：在Lambda环境里创建CSV文件的基础方法，以及结合Cost Explorer返回的数据生成对应报表。

一、Lambda中创建CSV的核心操作 #

Python标准库自带的csv模块就能完美搞定，而且Lambda的Python运行环境默认包含这个模块，不需要额外打包第三方库。基本步骤如下：

• 导入csv模块
• 整理好要写入的结构化数据（列表或字典格式都可以）
• 利用Lambda的临时目录/tmp创建文件（这是Lambda唯一允许写入的本地目录）
• 通过csv.writer或csv.DictWriter完成内容写入

举个极简示例：

import csv

def create_simple_csv():
    # 模拟数据
    data = [
        ["项目名称", "月度成本", "统计月份"],
        ["ProjectX", "156.70", "2024-03"],
        ["ProjectY", "92.45", "2024-03"]
    ]
    
    # Lambda临时文件路径
    csv_path = "/tmp/sample_report.csv"
    
    # 写入CSV
    with open(csv_path, 'w', newline='', encoding='utf-8') as csvfile:
        writer = csv.writer(csvfile)
        writer.writerows(data)
    
    return csv_path

二、结合Cost Explorer API结果生成业务报表 #

现在把你的Cost Explorer查询代码和CSV生成逻辑结合起来，核心是先解析API返回的response数据，提取出月份、Project标签、成本这些关键字段，再写入CSV。

完整可运行的Lambda代码： #

import boto3
import datetime
import re
import csv

def lambda_handler(event, context):
    client = boto3.client('ce')
    now = datetime.datetime.utcnow()
    end = datetime.datetime(year=now.year, month=now.month, day=1)
    start = end - datetime.timedelta(days=1)
    start = datetime.datetime(year=start.year, month=start.month, day=1)
    start_date = start.strftime('%Y-%m-%d')
    end_date = end.strftime('%Y-%m-%d')

    # 调用Cost Explorer API（这里用你动态计算的起止日期，替代原代码里的固定日期）
    response = client.get_cost_and_usage(
        TimePeriod={
            'Start': start_date,
            'End': end_date
        },
        Granularity='MONTHLY',
        Metrics=['BlendedCost'],
        GroupBy=[
            {
                'Type': 'TAG',
                'Key': 'Project'
            },
        ]
    )

    # 准备CSV表头和数据容器
    csv_header = ["统计月份", "Project标签", "混合成本(USD)"]
    csv_rows = [csv_header]

    # 解析API返回的结构化数据
    for time_result in response['ResultsByTime']:
        current_month = time_result['TimePeriod']['Start']
        # 遍历每个Project标签分组
        for group in time_result['Groups']:
            # 处理标签为空的情况，避免索引错误
            project_name = group['Keys'][0] if group['Keys'] else "未标记Project"
            # 提取成本数值
            blended_cost = group['Metrics']['BlendedCost']['Amount']
            # 添加到数据行
            csv_rows.append([current_month, project_name, blended_cost])

    # 第一步：写入Lambda临时文件
    csv_file_path = "/tmp/project_cost_report.csv"
    with open(csv_file_path, 'w', newline='', encoding='utf-8') as csvfile:
        writer = csv.writer(csvfile)
        writer.writerows(csv_rows)

    # 第二步：上传到S3永久存储（Lambda临时文件会在执行结束后被清理）
    # 注意：要给Lambda的IAM角色添加s3:PutObject权限
    s3_client = boto3.client('s3')
    s3_client.upload_file(
        csv_file_path,
        'your-s3-bucket-name',  # 替换成你的S3桶名称
        f"cost-reports/{datetime.datetime.utcnow().strftime('%Y%m%d_%H%M')}_project-cost.csv"
    )

    return {
        'statusCode': 200,
        'body': f"报表已生成并上传至S3，临时文件路径：{csv_file_path}"
    }

关键细节提醒： #

• 临时目录限制：Lambda只有/tmp目录可写，文件生命周期和函数执行周期绑定，所以必须上传到S3这类持久化存储才能保留报表。
• 权限配置：确保Lambda的IAM角色拥有ce:GetCostAndUsage（调用Cost Explorer）和s3:PutObject（上传S3）的权限。
• 异常处理：如果需要更健壮的代码，可以添加try-except块，处理API调用失败、S3上传失败等情况。
• 自定义输出：如果需要把CSV内容直接返回给调用方（比如API Gateway），可以用StringIO在内存中生成CSV，而不是写入本地文件。

内容的提问来源于stack exchange，提问作者mango5k