cloud-aws-databases

# 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, extensions - **MySQL**: Popular, good ecosystem - **Aurora**: AWS-optimized, 5x MySQL / 3x PostgreSQL performance - **MariaDB**: MySQL fork, additional features - **Oracle**: Commercial, enterprise features - **SQL Server**: Microsoft, Windows integration ```python 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', DBInstanceClass='db.t3.medium', # 2 vCPU, 4 GB RAM Engine='postgres', EngineVersion='15.4', MasterUsername='dbadmin', MasterUserPassword='SecurePassword123!', # Example only - use AWS Secrets Manager in production AllocatedStorage=100, # GB StorageType='gp3', # General purpose SSD StorageEncrypted=True, MultiAZ=True, # High availability DBSubnetGroupName='my-db-subnet-group', VpcSecurityGroupIds=['sg-0123456789abcdef0'], BackupRetentionPeriod=7, # Days PreferredBackupWindow='03:00-04:00', # UTC PreferredMaintenanceWindow='sun:04:00-sun:05:00', EnableCloudwatchLogsExports=['postgresql'], Tags=[ {'Key': 'Name', 'Value': 'myapp-db'}, {'Key': 'Environment', 'Value': 'production'} ] ) print(f"Creating RDS instance: {response['DBInstance']['DBInstanceIdentifier']}") return response['DBInstance']['Endpoint']['Address'] def create_read_replica(source_db_id): """Create read replica for scaling reads""" response = rds.create_db_instance_read_replica( DBInstanceIdentifier=f'{source_db_id}-replica-1', SourceDBInstanceIdentifier=source_db_id, DBInstanceClass='db.t3.medium', PubliclyAccessible=False, Tags=[ {'Key': 'Name', 'Value': f'{source_db_id}-replica'}, {'Key': 'Role', 'Value': 'read-replica'} ] ) print(f"Creating read replica: {response['DBInstance']['DBInstanceIdentifier']}") ``` ### Concept 2: DynamoDB **DynamoDB concepts**: - **Tables**: Primary key (partition + sort key) - **Indexes**: GSI (global), LSI (local) - **Capacity modes**: On-demand vs provisioned - **Streams**: Change data capture ```python 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'}, # Partition key ], 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'} ] ) # Wait for table to be created table.wait_until_exists() print(f"Created table: {table.table_name}") return table # CRUD operations 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 vs Memcached**: - **Redis**: Data structures, persistence, replication, pub/sub - **Memcached**: Simple key-value, multi-threaded, faster for simple caching ```python 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, # Primary + 1 replica 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']}") # Use Redis client def connect_to_redis(endpoint, port=6379): """Connect to ElastiCache Redis""" # For encrypted connections client = redis.Redis( host=endpoint, port=port, decode_responses=True, ssl=True, ssl_cert_reqs=None # For self-signed certs ) return client # Caching pattern def get_user_cached(user_id, redis_client): """Get user with Redis caching""" cache_key = f'user:{user_id}' # Check cache cached = redis_client.get(cache_key) if cached: return json.loads(cached) # Cache miss - fetch from DB user = get_user_from_db(user_id) # Store in cache (1 hour TTL) 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) ```python 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!', # Example only - use AWS Secrets Manager in production DatabaseName='myapp', DBSubnetGroupName='my-db-subnet-group', VpcSecurityGroupIds=['sg-0123456789abcdef0'], ServerlessV2ScalingConfiguration={ 'MinCapacity': 0.5, # ACUs (Aurora Capacity Units) 'MaxCapacity': 2.0 }, EnableHttpEndpoint=True, # Data API StorageEncrypted=True, BackupRetentionPeriod=7, Tags=[ {'Key': 'Name', 'Value': 'myapp-aurora'}, {'Key': 'Type', 'Value': 'serverless'} ] ) cluster_id = response['DBCluster']['DBClusterIdentifier'] # Create serverless instance 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 ```python import boto3 dms = boto3.client('dms') def create_dms_replication(): """Create DMS replication instance and task""" # Create replication instance 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 ) # Wait for instance to be available waiter = dms.get_waiter('replication_instance_available') waiter.wait( Filters=[ {'Name': 'replication-instance-id', 'Values': ['myapp-migration']} ] ) # Create source endpoint (on-premises database) 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' # Example only - use AWS Secrets Manager in production ) # Create target endpoint (RDS) 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!' # Example only - use AWS Secrets Manager in production ) # Create migration task 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', # Full load + ongoing replication TableMappings=json.dumps({ 'rules': [ { 'rule-type': 'selection', 'rule-id': '1',