[Python](備忘録)Boto3でCostExplorerからコストを取得する

AWSのコストはCostExprolerの画面から確認できますが、詳細な分析には機能が足りないと感じる時がありました。
PowerBIと連携し、より見やすく分析ができるようにCSV出力の機能を作ってみました。
※個人的なメモの為、ご利用なさる場合はご注意ください！

python main.py

ce = boto3.client('ce')
def get_cost_and_usage(**params):
    _response = ce.get_cost_and_usage(**params)

    _results = _response["ResultsByTime"]
    while "NextPageToken" in _response:
        params["NextPageToken"] = _response["NextPageToken"]
        _response = ce.get_cost_and_usage(**params)
        _results.extend(_response["ResultsByTime"])

    return _results

def make_cost_def(response):
    _dates , _names , _costs, _tags = [],[],[],[]

    for row in response:
        _date = row["TimePeriod"]["Start"]
        for group in row["Groups"]:
            _name = group["Keys"]
            _cost = group["Metrics"]["UnblendedCost"]["Amount"]
            if _cost == "0":
                # コスト計上に不要なデータは除外
                continue
            _dates.append(_date)
            _names.append(','.join(_name))
            _costs.append(_cost)
            _tags.append(tag_name)

    _result_df = pd.DataFrame(
        data={"date":_dates, "name":_names,"cost":_costs,"tag":_tags},
        columns=["date","name","cost","tag"])

    return _result_df

today = datetime.today()
tag_name = "hoge_name"
tag_value = "hoge_value"
search_at_months = 12
params = {
    'TimePeriod':{
        'Start': datetime.strftime(today - relativedelta(months=search_at_months), '%Y-%m-%d'),
        'End': datetime.strftime(today, '%Y-%m-%d')
    },
    'Granularity':'DAILY',
    'Metrics':['UnblendedCost'],
    'GroupBy':[{'Type': 'DIMENSION','Key': 'USAGE_TYPE'}],
    # Filter={'Tags': {'Key': tag_name,'Values': [tag_value]}}
}
make_cost_def(get_cost_and_usage(params)).to_csv("./cost.csv")

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

【おまけ】
上記出力結果をPowerBIで集計してみました。
各サービス毎の金額を簡単に見たりできるのでお勧めですよ！(個人利用のAWSコストなので少額なのはご愛敬^^;)

※EMRコストの調査ができるように、クラスター一覧を作成する処理も作成しました。

python main.py

import boto3
import pandas as pd
from datetime import datetime
from dateutil.relativedelta import relativedelta

emr = boto3.client('emr')

def get_list_clusters(**params):
    _response = emr.list_clusters(**params)
    _results = _response["Clusters"]
    while "Marker" in _response:
        params["Marker"] = _response["Marker"]
        _response = emr.get_cost_and_usage(**params)
        _results.extend(_response["Clusters"])

    return _results
    
def makeCulsterDef(response):
    _ids , _names , _normalized_instance_hours,_state_change_reasons, _creation_date_times,_end_date_times,_duration_times = [],[],[],[],[],[],[]

    for row in response:
        _ids.append(row["Id"])
        _names.append(row["Name"])
        _normalized_instance_hours.append(row["NormalizedInstanceHours"])
        _state_change_reasons.append(row["Status"]["StateChangeReason"]["Code"])
        _creation_date_times.append(row["Status"]["Timeline"]["CreationDateTime"])
        _end_date_times.append(row["Status"]["Timeline"]["EndDateTime"])
        _duration_times.append((row["Status"]["Timeline"]["EndDateTime"] - row["Status"]["Timeline"]["CreationDateTime"]).total_seconds())

    _result_df = pd.DataFrame(
        data={
            "id":_ids,
            "name":_names,
            "normalizedInstanceHours":_normalized_instance_hours,
            "stateChangeReasons":_state_change_reasons,
            "creationDateTimes":_creation_date_times,
            "endDateTimes":_end_date_times,
            "durationTimes":_duration_times},
        columns=["id","name","normalizedInstanceHours","stateChangeReasons","creationDateTimes","endDateTimes","durationTimes"])

    return _result_df

describe_params = {
    'CreatedAfter':datetime.strftime(today - relativedelta(months=1), '%Y-%m-%d'),
    'CreatedBefore':datetime.strftime(today, '%Y-%m-%d'),
}
make_culster_def(get_list_clusters(describe_params)).to_csv("./clusterList.csv")


#s3から設定ファイルのパスを取得
pattern = '^(17)/test/{1}[a-z,A-Z,0-9,-_]+/{1}[a-z,A-Z,0-9,-_]+/{1}application.properties'
def list_objects_v2(**params):
    _response = s3.list_objects_v2(**params)

    _results = _response["Contents"]
    while "ContinuationToken" in _response:
        params["ContinuationToken"] = _response["ContinuationToken"]
        _response = s3.list_objects_v2(**params)
        _results.extend(_response["Contents"])

    return _results

def makeDef(response):
    _types, _groups,_names = [],[],[]

    for row in response:
        _key = row["Key"]
        if re.match(pattern, _key):
            _types.append(_key.split("/")[0])
            _groups.append(_key.split("/")[1])
            _names.append(_key.split("/")[2])


    _result_df = pd.DataFrame(
        data={"type":_types, "group":_groups,"name":_names},
        columns=["type","group","name"])

    return _result_df

params = {
  "Bucket":"hoge-bucket",
  "Prefix":""
}
makeDef(list_objects_v2(**params)).to_csv("./clusterList.csv")