Skip to main content Skills Marktplatz Entdecken und erkunden Sie KI-Skills, die von der Community erstellt wurden.
Mit Codex oder Claude installieren Kopieren Sie diesen Prompt, fügen Sie ihn in Codex, Claude oder einen anderen Assistant ein und lassen Sie die Skill-Seite prüfen und installieren.
Prompt-Details anzeigen Ein direkter Befehl überspringt den Prüf-Prompt. Prüfen Sie die Quelle, bevor Sie ihn ausführen.
npx skills add https://github.com/tools-only/X-Skills --skill cloud-aws-databasesDer Befehl bleibt in einer Zeile. Scrollen Sie horizontal, um ihn vor dem Kopieren vollständig zu prüfen.
Sie bevorzugen eine lokale Kopie? Laden Sie die Dateien herunter, die SkillsMP derzeit vorliegen.
ZIP herunterladen Herunterladen... Mehr aus diesem Repository Verwandte Berufe SOC
Basierend auf der SOC-Berufsklassifikation
name cloud-aws-databases description AWS database services - RDS, DynamoDB, ElastiCache, Aurora, migration, backup, and optimization
AWS Databases
Scope : AWS databases - RDS, DynamoDB, ElastiCache, Aurora Serverless, migration strategies, backup and recovery
Lines : ~350
Last Updated : 2025-10-25
Format Version : 1.0 (Atomic)
When to Use This Skill
Activate this skill when:
Deploying managed relational databases with RDS
Building NoSQL applications with DynamoDB
Implementing caching with ElastiCache (Redis/Memcached)
Setting up Aurora Serverless for variable workloads
Migrating databases to AWS
Configuring database backups and point-in-time recovery
Optimizing database performance and read replicas
Troubleshooting database connectivity or performance issues
Core Concepts
Concept 1: RDS (Relational Database Service)
RDS engines :
PostgreSQL : Full-featured, JSON support, extensionsMySQL : Popular, good ecosystemAurora : AWS-optimized, 5x MySQL / 3x PostgreSQL performanceMariaDB : MySQL fork, additional featuresOracle : Commercial, enterprise featuresSQL Server : Microsoft, Windows integration
import boto3
rds = boto3.client('rds' )
def create_rds_instance ():
"""Create RDS PostgreSQL instance with Multi-AZ"""
response = rds.create_db_instance(
DBInstanceIdentifier='myapp-db'
'db.t3.medium'
'postgres'
'15.4'
'dbadmin'
'SecurePassword123!'
100
'gp3'
True
True
'my-db-subnet-group'
'sg-0123456789abcdef0'
7
'03:00-04:00'
'sun:04:00-sun:05:00'
'postgresql'
'Key'
'Name'
'Value'
'myapp-db'
'Key'
'Environment'
'Value'
'production'
print
f"Creating RDS instance: {response['DBInstance' ]['DBInstanceIdentifier' ]} "
return
'DBInstance'
'Endpoint'
'Address'
def
create_read_replica
source_db_id
"""Create read replica for scaling reads"""
f'{source_db_id} -replica-1'
'db.t3.medium'
False
'Key'
'Name'
'Value'
f'{source_db_id} -replica'
'Key'
'Role'
'Value'
'read-replica'
print
f"Creating read replica: {response['DBInstance' ]['DBInstanceIdentifier' ]} "
Concept 2: DynamoDB
Tables : Primary key (partition + sort key)Indexes : GSI (global), LSI (local)Capacity modes : On-demand vs provisionedStreams : Change data capture import boto3
from boto3.dynamodb.conditions import Key, Attr
from datetime import datetime
dynamodb = boto3.resource('dynamodb' )
def create_dynamodb_table ():
"""Create DynamoDB table with indexes"""
table = dynamodb.create_table(
TableName='Users' ,
KeySchema=[
{'AttributeName' : 'userId' , 'KeyType' : 'HASH' },
],
AttributeDefinitions=[
{'AttributeName' : 'userId' , 'AttributeType' : 'S' },
{'AttributeName' : 'email' , 'AttributeType' : 'S' },
{'AttributeName' : 'createdAt' , 'AttributeType' : 'S' }
],
GlobalSecondaryIndexes=[
{
'IndexName' : 'email-index' ,
'KeySchema' : [
{'AttributeName' : 'email' , 'KeyType' : 'HASH' }
],
'Projection' : {'ProjectionType' : 'ALL' },
'ProvisionedThroughput' : {
'ReadCapacityUnits' : 5 ,
'WriteCapacityUnits' : 5
}
},
{
'IndexName' : 'created-index' ,
'KeySchema' : [
{'AttributeName' : 'createdAt' , 'KeyType' : 'HASH' }
],
'Projection' : {'ProjectionType' : 'KEYS_ONLY' },
'ProvisionedThroughput' : {
'ReadCapacityUnits' : 5 ,
'WriteCapacityUnits' : 5
}
}
],
BillingMode='PROVISIONED' ,
ProvisionedThroughput={
'ReadCapacityUnits' : 10 ,
'WriteCapacityUnits' : 10
},
StreamSpecification={
'StreamEnabled' : True ,
'StreamViewType' : 'NEW_AND_OLD_IMAGES'
},
Tags=[
{'Key' : 'Environment' , 'Value' : 'production' }
]
)
table.wait_until_exists()
print (f"Created table: {table.table_name} " )
return table
table = dynamodb.Table('Users' )
def create_user (user_id, email, name ):
"""Create user item"""
table.put_item(
Item={
'userId' : user_id,
'email' : email,
'name' : name,
'createdAt' : datetime.utcnow().isoformat(),
'status' : 'active'
}
)
def get_user (user_id ):
"""Get user by ID"""
response = table.get_item(Key={'userId' : user_id})
return response.get('Item' )
def query_by_email (email ):
"""Query using GSI"""
response = table.query(
IndexName='email-index' ,
KeyConditionExpression=Key('email' ).eq(email)
)
return response['Items' ]
def update_user (user_id, name ):
"""Update user with atomic increment"""
response = table.update_item(
Key={'userId' : user_id},
UpdateExpression='SET #name = :name, updatedAt = :timestamp ADD loginCount :inc' ,
ExpressionAttributeNames={'#name' : 'name' },
ExpressionAttributeValues={
':name' : name,
':timestamp' : datetime.utcnow().isoformat(),
':inc' : 1
},
ReturnValues='ALL_NEW'
)
return response['Attributes' ]
Concept 3: ElastiCache
Redis : Data structures, persistence, replication, pub/subMemcached : Simple key-value, multi-threaded, faster for simple caching import boto3
import redis
elasticache = boto3.client('elasticache' )
def create_redis_cluster ():
"""Create ElastiCache Redis cluster"""
response = elasticache.create_replication_group(
ReplicationGroupId='myapp-redis' ,
ReplicationGroupDescription='Redis cluster for myapp' ,
Engine='redis' ,
EngineVersion='7.0' ,
CacheNodeType='cache.t3.medium' ,
NumCacheClusters=2 ,
AutomaticFailoverEnabled=True ,
MultiAZEnabled=True ,
CacheSubnetGroupName='my-cache-subnet-group' ,
SecurityGroupIds=['sg-0123456789abcdef0' ],
AtRestEncryptionEnabled=True ,
TransitEncryptionEnabled=True ,
SnapshotRetentionLimit=5 ,
SnapshotWindow='03:00-05:00' ,
Tags=[
{'Key' : 'Name' , 'Value' : 'myapp-redis' },
{'Key' : 'Environment' , 'Value' : 'production' }
]
)
print (f"Creating Redis cluster: {response['ReplicationGroup' ]['ReplicationGroupId' ]} " )
def connect_to_redis (endpoint, port=6379 ):
"""Connect to ElastiCache Redis"""
client = redis.Redis(
host=endpoint,
port=port,
decode_responses=True ,
ssl=True ,
ssl_cert_reqs=None
)
return client
def get_user_cached (user_id, redis_client ):
"""Get user with Redis caching"""
cache_key = f'user:{user_id} '
cached = redis_client.get(cache_key)
if cached:
return json.loads(cached)
user = get_user_from_db(user_id)
redis_client.setex(
cache_key,
3600 ,
json.dumps(user)
)
return user
Concept 4: Aurora Serverless Aurora Serverless use cases :
Variable workloads (dev/test environments)
Unpredictable traffic patterns
Multi-tenant applications
Infrequent usage (pauses when idle)
def create_aurora_serverless_cluster ():
"""Create Aurora Serverless v2 cluster"""
response = rds.create_db_cluster(
DBClusterIdentifier='myapp-aurora' ,
Engine='aurora-postgresql' ,
EngineVersion='15.4' ,
MasterUsername='dbadmin' ,
MasterUserPassword='SecurePassword123!' ,
DatabaseName='myapp' ,
DBSubnetGroupName='my-db-subnet-group' ,
VpcSecurityGroupIds=['sg-0123456789abcdef0' ],
ServerlessV2ScalingConfiguration={
'MinCapacity' : 0.5 ,
'MaxCapacity' : 2.0
},
EnableHttpEndpoint=True ,
StorageEncrypted=True ,
BackupRetentionPeriod=7 ,
Tags=[
{'Key' : 'Name' , 'Value' : 'myapp-aurora' },
{'Key' : 'Type' , 'Value' : 'serverless' }
]
)
cluster_id = response['DBCluster' ]['DBClusterIdentifier' ]
rds.create_db_instance(
DBInstanceIdentifier=f'{cluster_id} -instance-1' ,
DBInstanceClass='db.serverless' ,
Engine='aurora-postgresql' ,
DBClusterIdentifier=cluster_id
)
print (f"Created Aurora Serverless cluster: {cluster_id} " )
Patterns
Pattern 1: Database Migration with DMS When to use : Migrate databases to AWS with minimal downtime
import boto3
dms = boto3.client('dms' )
def create_dms_replication ():
"""Create DMS replication instance and task"""
replication_response = dms.create_replication_instance(
ReplicationInstanceIdentifier='myapp-migration' ,
ReplicationInstanceClass='dms.t3.medium' ,
AllocatedStorage=100 ,
VpcSecurityGroupIds=['sg-0123456789abcdef0' ],
MultiAZ=False ,
EngineVersion='3.4.7' ,
PubliclyAccessible=False
)
waiter = dms.get_waiter('replication_instance_available' )
waiter.wait(
Filters=[
{'Name' : 'replication-instance-id' , 'Values' : ['myapp-migration' ]}
]
)
source_endpoint = dms.create_endpoint(
EndpointIdentifier='source-postgres' ,
EndpointType='source' ,
EngineName='postgres' ,
ServerName='onprem-db.example.com' ,
Port=5432 ,
DatabaseName='myapp' ,
Username='migration_user' ,
Password='migration_password'
)
target_endpoint = dms.create_endpoint(
EndpointIdentifier='target-rds' ,
EndpointType='target' ,
EngineName='postgres' ,
ServerName='myapp-db.abc123.us-east-1.rds.amazonaws.com' ,
Port=5432 ,
DatabaseName='myapp' ,
Username='dbadmin' ,
Password='SecurePassword123!'
)
dms.create_replication_task(
ReplicationTaskIdentifier='myapp-full-load' ,
SourceEndpointArn=source_endpoint['Endpoint' ]['EndpointArn' ],
TargetEndpointArn=target_endpoint['Endpoint' ]['EndpointArn' ],
ReplicationInstanceArn=replication_response['ReplicationInstance' ]['ReplicationInstanceArn' ],
MigrationType='full-load-and-cdc' ,
TableMappings=json.dumps({
'rules' : [
{
'rule-type' : 'selection' ,
'rule-id' : '1' ,
'rule-name' : 'include-all' ,
'object-locator' : {
'schema-name' : 'public' ,
'table-name' : '%'
},
'rule-action' : 'include'
}
]
})
)
print ("Created DMS replication task" )
Pattern 2: Connection Pooling Use case : Manage database connections efficiently
import psycopg2
from psycopg2 import pool
db_pool = psycopg2.pool.SimpleConnectionPool(
minconn=1 ,
maxconn=20 ,
host='myapp-db.abc123.us-east-1.rds.amazonaws.com' ,
port=5432 ,
database='myapp' ,
user='dbadmin' ,
password='SecurePassword123!'
)
def execute_query (query, params=None ):
"""Execute query using connection from pool"""
conn = None
try :
conn = db_pool.getconn()
cursor = conn.cursor()
cursor.execute(query, params)
if cursor.description:
results = cursor.fetchall()
else :
results = None
conn.commit()
return results
except Exception as e:
if conn:
conn.rollback()
raise e
finally :
if conn:
db_pool.putconn(conn)
connection_pool = None
def lambda_handler (event, context ):
"""Lambda with persistent connection pool"""
global connection_pool
if not connection_pool:
connection_pool = create_connection_pool()
results = execute_query_pooled(connection_pool, "SELECT * FROM users LIMIT 10" )
return {
'statusCode' : 200 ,
'body' : json.dumps({'users' : results})
}
Pattern 3: DynamoDB Batch Operations Use case : Efficient bulk reads/writes
def batch_write_items (table_name, items ):
"""Batch write up to 25 items at a time"""
dynamodb = boto3.resource('dynamodb' )
table = dynamodb.Table(table_name)
with table.batch_writer() as batch:
for item in items:
batch.put_item(Item=item)
print (f"Batch wrote {len (items)} items" )
def batch_get_items (table_name, keys ):
"""Batch get up to 100 items at a time"""
dynamodb = boto3.resource('dynamodb' )
response = dynamodb.batch_get_item(
RequestItems={
table_name: {
'Keys' : keys,
'ConsistentRead' : True
}
}
)
items = response['Responses' ][table_name]
while response.get('UnprocessedKeys' ):
response = dynamodb.batch_get_item(
RequestItems=response['UnprocessedKeys' ]
)
items.extend(response['Responses' ][table_name])
return items
Pattern 4: Database Backup and Restore Use case : Point-in-time recovery and snapshots
def create_rds_snapshot (db_instance_id ):
"""Create manual snapshot"""
snapshot_id = f"{db_instance_id} -{datetime.utcnow().strftime('%Y%m%d-%H%M%S' )} "
response = rds.create_db_snapshot(
DBSnapshotIdentifier=snapshot_id,
DBInstanceIdentifier=db_instance_id,
Tags=[
{'Key' : 'Type' , 'Value' : 'manual' },
{'Key' : 'CreatedBy' , 'Value' : 'automation' }
]
)
print (f"Creating snapshot: {snapshot_id} " )
return snapshot_id
def restore_from_snapshot (snapshot_id, new_instance_id ):
"""Restore database from snapshot"""
response = rds.restore_db_instance_from_db_snapshot(
DBInstanceIdentifier=new_instance_id,
DBSnapshotIdentifier=snapshot_id,
DBInstanceClass='db.t3.medium' ,
PubliclyAccessible=False ,
MultiAZ=True
)
print (f"Restoring {new_instance_id} from {snapshot_id} " )
def point_in_time_restore (source_db_id, target_db_id, restore_time ):
"""Restore to specific point in time"""
response = rds.restore_db_instance_to_point_in_time(
SourceDBInstanceIdentifier=source_db_id,
TargetDBInstanceIdentifier=target_db_id,
RestoreTime=restore_time,
DBInstanceClass='db.t3.medium'
)
print (f"Restoring {target_db_id} to {restore_time} " )
Quick Reference
Database Service Selection Use Case Service Type Best For Relational, ACID RDS SQL Structured data, transactions Key-value, high scale DynamoDB NoSQL Serverless, millisecond latency Caching, sessions ElastiCache In-memory Performance optimization Variable workload Aurora Serverless SQL Cost optimization Graph data Neptune Graph Relationships, social networks Time series Timestream Time series IoT, metrics, logs
RDS Instance Sizing Workload Type | Instance Class | Example | vCPU | RAM
-------------------|----------------|---------------|------|-------
Dev/test | db.t3.micro | db.t3.micro | 2 | 1 GB
Small production | db.t3.medium | db.t3.medium | 2 | 4 GB
Medium production | db.m5.large | db.m5.large | 2 | 8 GB
Large production | db.r5.xlarge | db.r5.xlarge | 4 | 32 GB
Memory-intensive | db.r5.4xlarge | db.r5.4xlarge | 16 | 128 GB
Key Guidelines ✅ DO: Enable Multi-AZ for production RDS instances
✅ DO: Use read replicas to scale read traffic
✅ DO: Enable automated backups (7-35 days retention)
✅ DO: Use connection pooling for Lambda functions
✅ DO: Enable encryption at rest and in transit
✅ DO: Use IAM database authentication when possible
✅ DO: Monitor performance with CloudWatch
❌ DON'T: Use DynamoDB scans for large tables (use queries)
❌ DON'T: Expose databases publicly (use VPC endpoints)
❌ DON'T: Ignore read replica lag for critical queries
❌ DON'T: Use provisioned capacity without monitoring
❌ DON'T: Store large objects in DynamoDB (use S3 + pointers)
Anti-Patterns
Critical Violations
def lambda_handler (event, context ):
conn = psycopg2.connect(
host='db.example.com' ,
database='myapp' ,
user='dbadmin' ,
password='password'
)
connection = None
def lambda_handler (event, context ):
global connection
if not connection or connection.closed:
connection = psycopg2.connect(...)
cursor = connection.cursor()
cursor.execute("SELECT * FROM users" )
❌ New connection per invocation : Exhausts database connections, high latency
✅ Correct approach : Initialize outside handler, reuse across warm invocations
Common Mistakes
response = table.scan()
items = response['Items' ]
response = table.query(
IndexName='email-index' ,
KeyConditionExpression=Key('email' ).eq(user_email)
)
items = response['Items' ]
❌ DynamoDB scan : High latency, expensive, consumes capacity
✅ Better : Use query with partition key, add GSI if needed
Related Skills
aws-lambda-functions.md - Lambda integration with databasesaws-storage.md - S3 for database backups and large objectsaws-networking.md - VPC endpoints for private database accessaws-iam-security.md - IAM database authentication and permissions
Last Updated : 2025-10-25
Format Version : 1.0 (Atomic)
