Skip to content

Instantly share code, notes, and snippets.

@nukemberg
Created June 22, 2021 16:24
Show Gist options
  • Save nukemberg/27a7de159d8434bfa489f6f9d8a8b074 to your computer and use it in GitHub Desktop.
Save nukemberg/27a7de159d8434bfa489f6f9d8a8b074 to your computer and use it in GitHub Desktop.
DynamoDB table WCU/RCU reservation calculator
import boto3
from datetime import datetime
from dateutil.relativedelta import relativedelta
import numpy as np
def get_table_metrics(cloudwatch_client, table_name, start_time, end_time):
response = cloudwatch_client.get_metric_data(
MetricDataQueries=[
{
'Id': 'dynamoDBWCU',
'MetricStat': {
'Metric': {
'Namespace': 'AWS/DynamoDB',
'MetricName': 'ConsumedWriteCapacityUnits',
'Dimensions': [
{'Name': 'TableName', 'Value': table_name}
]
},
'Period': 3600,
'Stat': 'Sum'
}
},
{
'Id': 'dynamoDBRCU',
'MetricStat': {
'Metric': {
'Namespace': 'AWS/DynamoDB',
'MetricName': 'ConsumedReadCapacityUnits',
'Dimensions': [
{'Name': 'TableName', 'Value': table_name}
]
},
'Period': 3600,
'Stat': 'Sum'
}
}
],
StartTime=start_time,
EndTime=end_time
)
rcu_data = next(filter(lambda r: r['Id'] == 'dynamoDBRCU', response['MetricDataResults']))
wcu_data = next(filter(lambda r: r['Id'] == 'dynamoDBWCU', response['MetricDataResults']))
return (rcu_data['Values'], wcu_data['Values'])
# NOTE: you must have at least one month (preferably more) of data for this to be useful
# modify this for your needs
table_name = 'SOME_TABLE'
end_time = datetime.now()
start_time = end_time - relativedelta(years=1)
cloudwatch = boto3.client('cloudwatch')
rcu_data, wcu_data = get_table_metrics(cloudwatch, table_name, start_time, end_time)
print('WCU reservations')
print('With partial usage: {}, no partial usage: {}'.format(*np.percentile(wcu_data, [32, 46])))
print('RCU reservations')
print('With partial usage: {}, no partial usage: {}'.format(*np.percentile(rcu_data, [32, 46])))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment