AWS Cost and Usage Report与账单控制台成本差异排查求助

我已经激活了按小时生成的AWS成本与使用情况报告（CUR），格式为Parquet。用Python读取该Parquet文件计算成本时，发现结果和AWS账单控制台显示的差异极大——比如EC2在控制台显示约12000美元，但CUR计算结果只有约4500美元。以下是我的Python代码，使用的是最新的Parquet文件：

import sys
import time
from fastparquet import ParquetFile
import json
import datetime
from collections import defaultdict

tags_key = 'resource_tags_'
def read_parquet_file(parquet_file_path):
    start_time_rf = time.time()
    dataset = ParquetFile(parquet_file_path)

    # Convert the Parquet table to a Pandas DataFrame 
    df = dataset.to_pandas()

    resource_tags_columns = [col for col in df.columns if col.startswith(tags_key)]
    # print(resource_tags_columns)
    
    relevant_columns = ['identity_line_item_id', 'product_sku', 'line_item_line_item_type', 
                        'line_item_resource_id', 'line_item_product_code', 'product_region', 
                        'bill_billing_period_start_date', 'bill_billing_period_end_date',
                        'line_item_usage_start_date', 'line_item_usage_end_date',
                        'product_usagetype'] + resource_tags_columns + ['line_item_unblended_cost']
    df = df[relevant_columns]

    # Perform the aggregation directly without grouping if possible
    grouped_data = df.groupby(relevant_columns[:-1])['line_item_unblended_cost'].sum()
    # Sort the grouped data by line_item_product_code
    grouped_data = grouped_data.sort_index(level='line_item_product_code')
     # Record the end time
    end_time_rf = time.time()

    # Calculate the elapsed time
    elapsed_time = end_time_rf - start_time_rf
    # print(f'Time taken to read the Parquet file: {elapsed_time} seconds')
    return grouped_data, resource_tags_columns

def parse_data(grouped_data, resource_tags_columns):
    start_time_rf = time.time()
    # Print the total cost for each 'identity_line_item_id'
    final_sum = 0
    tax = 0
    usage = 0

    # grouping data based in username
    final_grouped_data = defaultdict(list)
    def convert_cost(cost):
        try:
            # Attempt direct string formatting for potential efficiency
            return f"{cost:.{len(str(cost).split('.')[-1])}f}"
        except ValueError:
            # Handle cases where direct formatting fails (e.g., scientific notation)
            exponent_str = str(cost)
            base, power = exponent_str.split('e')
            decimal = len(base.split('.')[1]) if len(base.split('.')) > 1 else 0
            power = abs(int(power))
            return f"{cost:.{power + decimal}f}"

    for (identity_line_item_id,product_sku,line_item_line_item_type,line_item_resource_id,line_item_product_code, 
        product_region ,bill_billing_period_start_date,
        bill_billing_period_end_date, line_item_usage_start_date, line_item_usage_end_date,
        product_usagetype, *resource_tags),total_cost in grouped_data.items():
        final_sum += total_cost
        if (line_item_line_item_type == 'Tax' ):
            tax += total_cost
        elif (line_item_line_item_type == 'Usage'):
            usage += total_cost

        entry = {
            'identityLineItemId': identity_line_item_id,
            'line_item_resource_id': line_item_resource_id,
            'totalCost': convert_cost(total_cost),
            'product_sku': product_sku,
            'region': product_region,
            'line_item_line_item_type': line_item_line_item_type,
            'line_item_product_code':line_item_product_code,
            'product_usagetype': product_usagetype,
            'start_date': datetime.datetime.strftime(bill_billing_period_start_date, '%Y-%m-%d %H:%M:%S'),
            'end_date':  datetime.datetime.strftime(bill_billing_period_end_date,'%Y-%m-%d %H:%M:%S'),
            'usage_start_date':datetime.datetime.strftime(line_item_usage_start_date, '%Y-%m-%d %H:%M:%S'),
            'usage_end_date':datetime.datetime.strftime(line_item_usage_end_date, '%Y-%m-%d %H:%M:%S'),
        }

        final_grouped_data[line_item_product_code].append(entry)

        for col, tag_key in enumerate(resource_tags_columns):
            if tag_value := resource_tags[col]:
                entry[tag_key] = tag_value
        

    def groupByRegion(data):
        group_by_region = defaultdict(list)  # Use defaultdict for efficient handling of new keys
        for entry in data:
            if region := entry.get('region'):
                group_by_region[region].append(entry)

        return group_by_region

    def groupByIdentityLineItemId(data):
        obj = {}
        for entry in data:
            # Extract values
            identity_id = entry["identityLineItemId"]
            total_cost = float(entry["totalCost"])

            # Add to existing entry or create a new one
            if identity_id in obj:
                obj[identity_id]["total_cost"] = convert_cost(float(obj[identity_id]["total_cost"]) + total_cost)
            else:
                obj[identity_id] = {
                    "total_cost": convert_cost(total_cost),
                    "identityLineItemId": entry['identityLineItemId'],
                    "line_item_resource_id": entry['line_item_resource_id'],
                    "region": entry['region'],
                    "line_item_product_code": entry['line_item_product_code'],
                    "product_usagetype": entry['product_usagetype']
                }
                for key, value in entry.items():
                    if key.startswith(tags_key):
                        obj[identity_id][key] = value
        return obj
    
    def groupByProductUsageType(data, type):
        group_by_type = defaultdict(list)  # Use defaultdict for efficient handling of new keys
        for entry in data:
            if type == 'AmazonEC2':
                if 'EBS' in entry['product_usagetype']:
                    group_by_type['EBS'].append(entry)
                elif 'AmazonEC2Stopped' in entry['product_usagetype']:
                    group_by_type['AmazonEC2Stopped'].append(entry)
                else:
                    group_by_type['AmazonEC2Running'].append(entry)
                # group_by_type['EBS' if 'EBS' in entry['product_usagetype'] else 'AmazonEC2Running'].append(entry)
            else:
                group_by_type[entry['product_usagetype']].append(entry)
        return getCostByproductUsageType(group_by_type)

    def getCostByRegion(data, type):
        group_by_region = defaultdict(list)  # Use defaultdict for efficient handling
        for region, entries in data.items():
            region_wise_sum = sum(float(entry['totalCost']) for entry in entries)  # Sum total cost using generator expression
            details = groupByProductUsageType(entries, type)  # Group by product usage type
            usage_sum = 0
            for entry in entries:
                if entry['line_item_line_item_type'] == 'Usage':
                    usage_sum += float(entry['totalCost'])
            
            group_by_region[region].append({'total_cost': convert_cost(region_wise_sum),
                'usage_cost': convert_cost(usage_sum), 'details': details})
        return group_by_region

    def getCostByproductUsageType(data):
        group_by_type = defaultdict(list)
        for key, value in data.items():
            regionWiseSum = sum(float(entry['totalCost']) for entry in value)  # Efficient summation

            # Group and convert cost within the loop for clarity
            group_by_type[key].append({
                f'{key}_total_cost': convert_cost(regionWiseSum),
                f'{key}_details': groupByIdentityLineItemId(value)
            })
        return group_by_type


        
    def get_cost_each_line_item_product(grouped_data):
        cost_details = {
            'details': {},
            'cost_details': {
                'total_cost': f'{final_sum:.2f}',
                'tax_cost': f'{tax:.2f}',
                'usage_cost': f'{usage:.2f}'
            }
        }

        for key, data in grouped_data.items():
            total_cost = sum(float(entry['totalCost']) for entry in data)
            tax_service = sum(float(entry['totalCost']) for entry in data if entry['line_item_line_item_type'] == 'Tax')
            usage_service = sum(float(entry['totalCost']) for entry in data if entry['line_item_line_item_type'] == 'Usage')
            region_wise_details = groupByRegion(data) 
            cost_details['details'][key] = {
                'total_cost': convert_cost(total_cost),
                'usage_cost': convert_cost(usage_service),
                'tax_cost': convert_cost(tax_service),
                'region_wise_details': getCostByRegion(region_wise_details, key)
            }

        formatted_json = json.dumps(cost_details, indent=2, ensure_ascii=False)

        # with open('final_code_cpy.json', 'w') as json_file:
        #     json.dump(cost_details, json_file)

        return formatted_json

    data = get_cost_each_line_item_product(final_grouped_data)
    print(data)
    # Record the end time
    end_time_rf = time.time()
    elapsed_time = end_time_rf - start_time_rf
    print(f'Time taken to scan the Parquet file: {elapsed_time} seconds')


if __name__ == '__main__':
   if len(sys.argv) < 2:
       sys.exit(1)

   parquet_file_path = sys.argv[1]
   grouped_data, resource_tags_columns = read_parquet_file(parquet_file_path)
   parse_data(grouped_data, resource_tags_columns)

注意：我使用的是最新的Parquet文件。

问题排查与修复方案 #

1. 成本字段选择错误 #

你当前仅使用line_item_unblended_cost，但AWS控制台显示的是抵扣后的最终费用，而unblended_cost是未应用预留实例、Savings Plans等折扣的原始费用。需要替换为以下字段：

• line_item_blended_cost：包含预留实例、Savings Plans抵扣后的费用，更接近控制台显示值
• line_item_total_cost：包含所有调整项（折扣、退款、信用额度）后的最终费用

修复：修改代码中聚合的成本字段，比如把line_item_unblended_cost替换为line_item_blended_cost。

2. 遗漏关键行项类型 #

你的代码只统计了Usage和Tax类型的行项，但CUR中还有其他影响总成本的类型：

• Discount：折扣抵扣金额
• Credit：信用额度抵扣
• RIFee：预留实例费用
• SavingsPlansCoveredUsage：Savings Plans覆盖的费用

这些项会直接影响总成本统计，遗漏会导致计算结果偏低。

修复：去掉行项类型过滤，直接累加所有类型的成本；如果需要区分类型，保留所有类型的统计逻辑。

3. EC2费用拆分问题 #

控制台的EC2总费用包含实例运行费、EBS存储费、数据传输费等，但你的代码把EBS单独分组，且部分EBS行项的product_code是AmazonEBS，未纳入EC2统计。

修复：统计EC2总费用时，同时包含AmazonEC2和AmazonEBS的相关行项。

4. 重复分组导致精度丢失 #

代码中先按多列分组求和，之后又多次将浮点型成本转为字符串再转回浮点，容易出现精度丢失或重复计算问题。

修复：直接在DataFrame上完成所有聚合操作，保留浮点型成本直到最终输出，避免中途类型转换。

简化后的验证代码 #

import sys
import pandas as pd
import json

def calculate_aws_cost(parquet_path):
    # 读取Parquet文件
    df = pd.read_parquet(parquet_path)
    
    # 选择关键成本字段
    cost_cols = [
        'line_item_product_code', 'line_item_line_item_type',
        'line_item_blended_cost', 'product_region'
    ]
    df = df[cost_cols].dropna(subset=['line_item_blended_cost'])
    
    # 计算总成本
    total_cost = df['line_item_blended_cost'].sum()
    
    # 计算EC2包含EBS的总费用
    ec2_ebs_cost = df[
        df['line_item_product_code'].isin(['AmazonEC2', 'AmazonEBS'])
    ]['line_item_blended_cost'].sum()
    
    # 按产品分组统计
    product_cost = df.groupby('line_item_product_code')['line_item_blended_cost'].sum().reset_index()
    
    # 输出结果
    result = {
        'total_cost': round(float(total_cost), 2),
        'ec2_including_ebs_cost': round(float(ec2_ebs_cost), 2),
        'product_breakdown': product_cost.to_dict('records')
    }
    print(json.dumps(result, indent=2))

if __name__ == '__main__':
    if len(sys.argv) < 2:
        sys.exit(1)
    calculate_aws_cost(sys.argv[1])

内容的提问来源于stack exchange，提问作者Not A Robot