如何汇总 AWS 账单中各项服务的 Data Transfer 费用

账单一直是客户在使用 AWS 过程中最重要的功能之一，为了能够和财务接口的自动化对接，AWS 推出了 AWS 成本管理器，其中丰富的 API 能够满足客户的自定义查询分析的同时又能够激发客户更多有想象力的需求。比如，借助 AWS 强大的无服务化服务如 Lambda，SQS，SNS 实现账单的自动拆分以及自动推送。再比如，借助成本管理器 API 以及 Pandas，NumPy 等数据处理工具，可以生成各种维度以及不同格式的数据和上游的财务系统无缝对接。本博文以实际客户在 AWS 账单中遇到的挑战为例，详细介绍了如何使用 AWS 成本管理器 API 自动生成 AWS 标准账单，并提供了可借鉴的代码供有兴趣的来者进一步挖掘实现深层次的需求以及功能。

背景

AWS Cost Explorer 成本管理服务的界面易于使用，可让客户直观看到、理解和管理随着时间变化的 AWS 成本和使用情况。越来越多的客户使用 AWS Cost Explorer 结合 AWS 控制台账单来分析在各个 Region 以及服务的具体费用分布。同时，数据传输费用一直是 AWS 账单的重要组成部分，尤其是当部署在AWS中的工作负载时直接面对终端用户时，数据传输费用往往在整个账单中占有很大的比重，因此这部分费用往往客户非常关注的一部分内容。

在现有的 AWS 账单系统中，为了给客户一个直观的印象，将各个服务产生的传输费用统一整合成为单列一项 “Data Transfer”，如下图所示：

然而，在通过 AWS Cost Explorer 成本管理器读取数据中，又将传输费用分担到各个服务中，如下图所示：

由于两者单列项显示的不同，虽然账单总额和 AWS 成本管理器中读取的细类能够保证统一，但是系列分项无法对齐。当客户希望实现 AWS 账单系统的自动化管理时，只能使用 AWS 成本管理器提供的 API 提取底层数据，而成本管理器系列分项和账单的显示不一致往往给客户的财务以及运维人员带来很大的困惑。同时由于这种显示的不一致也给财务人员自动化入账带来很多的麻烦。

为了解决控制台账单和 AWS Cost Explorer 不能匹配，本文使用 AWS 成本管理器 API 直接从 AWS 费用管理器接口取得全部费用数据，根据 AWS 控制台账单中 “Data Transfer”的计算逻辑，使用 Pandas 以及 NumPy 工具进行计算，汇总出和 AWS 控制台账单一致的数据，供客户财务系统使用。同时，从直观上使 AWS 成本管理器输出的数据和AWS控制台账单数据保持一致，避免运维人员反复和财务人员解释 “Data Transfer”费用在账单中生成的逻辑，节约了时间，达到了事半功倍的效果。本文中出现的例子代码，直接将 AWS 费用管理器中取出的数据生成 JSON 格式的输出文件，可以直接被用友财务软件读取。

实现思路

具体代码实现过程

客户需求

综上客户需求，能够使用 AWS 成本管理器 API 取出和账单完全一致的按照区域和服务类型【包含 Data Transfer 项】的费用详细列表以便和财务系统进行整合。输出格式要求 JSON 格式。

API 操作限制条件

通过阅读 AWS 成本管理器 API 文档如下，已知使用 get_cost_and_usage 提取详细数据的 API 有如下限制，【这里以基于 Python 的 Boto3 为例，其他语言类似】，当使用 GroupBy 进行数据提取时，只能进行两组分组，即 GroupBy 后分组条件最多只允许两个维度进行组合。

https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/ce/client/get_cost_and_usage.html

代码实现思路

根据 AWS 账单计费逻辑，Usage Type 为账单计费的基本逻辑单位。使用 AWS 成本管理器 API 获得详细数据，构造如下数据结构，然后通过 Usage Type 中属于 Data transfer 的类型进行重新汇总，生成相应的 Data transfer 服务类型，和 AWS 账单进行匹配。

1. 导入 Boto3 以及使用 AWS AK-SK profile 设置运行环境如下：

# 指定AWS configure profile
session = boto3.Session(profile_name='jingamz')
aws_client = session.client('ce')
# 指定查询范围，注意，[start,end)
start = '2023-01-01'
end = '2023-02-01'

2. 使用 get_cost_and_usage API 读取 AWS 成本管理器中的指定时间范围的账单费用数据。

def getcedetail(aws_client, startdate, enddate, Dimension1, Dimension2):
    start = startdate
    end = enddate
    DimKey1 = Dimension1
    DimKey2 = Dimension2
    ce = aws_client
    response = ce.get_cost_and_usage(
        TimePeriod={
            'Start': start,
            'End': end,
        },
        Granularity='MONTHLY',
        Metrics=[
            'UnblendedCost'
        ],
        GroupBy=[
            {
                'Type': 'DIMENSION',
                'Key': DimKey1
            },
            {
                'Type': 'DIMENSION',
                'Key': DimKey2
            }
        ]
    )
    return response['ResultsByTime']

3. 将输出的 Json 结果构造成为可供 pandas 进行组合查询的数据集合。

def GenCeReport(QueryOutput, merged_cost, dimkey1):
    for index, item in enumerate(QueryOutput):
        normalized_json = pandas.json_normalize(item['Groups'])
        split_keys = pandas.DataFrame(
            normalized_json['Keys'].tolist(),
            columns=[dimkey1, 'UsageType']
        )
        cost = pandas.concat(
            [split_keys, normalized_json['Metrics.UnblendedCost.Amount']],
            axis=1
        )
        renamed_cost = cost.rename(
            columns={'Metrics.UnblendedCost.Amount': item['TimePeriod']['Start']}
        )
        merged_cost = pandas.merge(merged_cost, renamed_cost, on=[dimkey1, 'UsageType'], how='right')
    return merged_cost

4. 在主函数中分别按照不同维度查询出，并且以 Usage type 作为关联进行数据合并。

# 构造DF
merged_cost_service = condf('Services', 'UsageType')
# 以 services 和 UsageType 为维度出数据。
response_service = getcedetail(aws_client, start, end, 'SERVICE', 'USAGE_TYPE')
# 生成DF
cost_service = GenCeReport(response_service, merged_cost_service, 'Services')

merged_cost_region = condf('Region', 'UsageType')
response_region = getcedetail(aws_client, start, end, 'REGION', 'USAGE_TYPE')
cost_region_filter = GenCeReport(response_region, merged_cost_region, 'Region')
# 360 filter NoRegion
cost_region = cost_region_filter.query("Region != 'NoRegion'")

# 以 UsageType 为轴进行合并
MyRegion = pandas.merge(cost_service, cost_region.drop([start], axis=1), on=['UsageType'],how='left')

5. 将对应的 Usage Type 转换成为新的 Data Transfer 类型，同时将 EC2-other 也合并到 EC2 中：

# 合并生成DataTransfer 类型的数据
MyRegion['Services'] = np.where((MyRegion['UsageType'].str.contains('DataTransfer-Out-Bytes')) |
                                (MyRegion['UsageType'].str.contains('DataTransfer-In-Bytes')) |
                                (MyRegion['UsageType'].str.contains('AWS-In-Bytes')) |
                                (MyRegion['UsageType'].str.contains('AWS-Out-Bytes')) |
                                (MyRegion['UsageType'].str.contains('DataXfer-In')) |
                                (MyRegion['UsageType'].str.contains('DataXfer-Out')) |
                                (MyRegion['UsageType'].str.contains('CloudFront-In-Bytes')) |
                                (MyRegion['UsageType'].str.contains('CloudFront-Out-Bytes')) |
                                (MyRegion['UsageType'].str.contains('DataTransfer-Regional-Bytes'))
                                , 'DataTransfer', MyRegion['Services'])
# 合并 'EC2 - Other' 到 'Amazon Elastic Compute Cloud - Compute'
MyRegion['Services'] = np.where((MyRegion['Services'].str.contains('EC2 - Other')),
                                'Amazon Elastic Compute Cloud - Compute', MyRegion['Services'])

6. 将生成的账单数据按照透视表的方式进行聚合，同时在存储成为中间态文件：

MyRegion.to_csv('/tmp/add-region.csv', index=True)
df = pandas.read_csv('/tmp/add-region.csv')
df.head()
bill_pivot = pandas.pivot_table(df, index=['Services', 'Region'], values=[start], aggfunc=np.sum, margins=True)

bill_pivot.to_csv('/tmp/result.csv')
df1 = pandas.read_csv('/tmp/result.csv')
usage = df1.rename(columns={start: 'Usage'})
list1 = []

7. 将生成的最终数据集转换成为 Json 格式供财务接口直接调用：

for row in usage.itertuples():
    dict_row = {
        "Keys": [
            row.Services,
            row.Region
        ],
        "Metrics": {
            "UnblendedCost": {
                "Amount": row.Usage,
                "Unit": "USD"
            }
        }
    }
    list1.append(dict_row)

dict_result = {
    "TimePeriod": {
        "Start": start,
        "End": end
    },
    "Total": {},
    "Groups": list1
}

# 注意这里需要替换放置输出的文件。
jsonFile = open('/Users/lijing/Desktop/CE20230301.json', 'w')
jsonFile.write(json.dumps(dict_result, indent=4))
jsonFile.close()

8. 最终代码在测试账号中运行结果如下：

方案成本

每个 API 调用花费 0.01 USD。从这个角度，如果为财务接口服务，每月仅需调用 API 调用两次就能完成月度账单的整合，总计调用费用 0.02 USD/月，折合人民币每月 0.15¥ 左右。其成本几乎可以忽略不计。

总结

本文介绍了如何调用 AWS 成本管理器 API 获取数据，通过 Pandas 以及 Numpy 数据处理工具生成可供财务接口调用的 Json 格式数据。通过这种方式加大增加了 AWS 账单和财务系统的对接的灵活性，仅需少量代码就能根据实际需要实现和任何类型财务接口的对接。相对适用 AWS CUR 进行数据汇总，无论从成本的角度还是操作便捷的角度，都能大大减少客户运维人员的压力。同样的其核心代码也可以供 AWS Lambda 调用，结合 AWS SQS 以及 SNS 服务实现费用账单的自动分配和定向邮件发送，达到事半功倍的效果。

附上参考 Python 代码：

https://github.com/lijingfz/CostExplorer/blob/main/CeTeat360Jingamz.py

亚马逊AWS官方博客