如何使用boto3筛选WCU/RCU≥200的DynamoDB表?
解决方案
直接给你能跑的代码,遍历所有表筛选出符合条件的:
import boto3 def get_all_table_names(): # 处理分页,获取所有表名(解决超过100张表的情况) dynamodb = boto3.client('dynamodb') table_names = [] last_evaluated_table = None while True: if last_evaluated_table: response = dynamodb.list_tables(ExclusiveStartTableName=last_evaluated_table) else: response = dynamodb.list_tables() table_names.extend(response['TableNames']) last_evaluated_table = response.get('LastEvaluatedTableName') if not last_evaluated_table: break return table_names def find_high_throughput_tables(): dynamodb = boto3.client('dynamodb') tables = get_all_table_names() high_throughput_tables = [] for table_name in tables: # 获取表的完整配置信息 table_desc = dynamodb.describe_table(TableName=table_name) table_data = table_desc['Table'] # 读取计费模式,默认是预置模式 billing_mode = table_data.get('BillingModeSummary', {}).get('BillingMode', 'PROVISIONED') # 只有预置计费模式的表才有固定的RCU/WCU,按需模式跳过 if billing_mode == 'PROVISIONED': provisioned = table_data['ProvisionedThroughput'] rcu = provisioned['ReadCapacityUnits'] wcu = provisioned['WriteCapacityUnits'] # 筛选RCU或WCU≥200的表 if rcu >= 200 or wcu >= 200: high_throughput_tables.append({ 'TableName': table_name, 'ReadCapacityUnits': rcu, 'WriteCapacityUnits': wcu }) # 输出结果 if high_throughput_tables: print("符合条件的表:") for table in high_throughput_tables: print(f"表名: {table['TableName']}, RCU: {table['ReadCapacityUnits']}, WCU: {table['WriteCapacityUnits']}") else: print("没有找到RCU/WCU≥200的表") if __name__ == "__main__": find_high_throughput_tables()
关键说明:
- 用
boto3.client比resource更适合批量遍历操作,效率更高 - 加了分页逻辑,能处理你200多张表的场景(默认
list_tables一次最多返回100张) - 按需计费模式的表没有固定的RCU/WCU配置,所以直接跳过这类表
- 只筛选预置模式下RCU或WCU≥200的表,把表名和对应数值统一输出
注意:
确保你的AWS凭证拥有dynamodb:ListTables和dynamodb:DescribeTable权限,否则会报错。
内容的提问来源于stack exchange,提问作者vineet singh
相关产品推荐
相关产品推荐

