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

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

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

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)

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

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',
                    '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

# Create connection 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!'  # Example only - use AWS Secrets Manager in production
)

def execute_query(query, params=None):
    """Execute query using connection from pool"""

    conn = None
    try:
        # Get connection from pool
        conn = db_pool.getconn()
        cursor = conn.cursor()

        # Execute query
        cursor.execute(query, params)

        # Fetch results
        if cursor.description:
            results = cursor.fetchall()
        else:
            results = None

        # Commit transaction
        conn.commit()

        return results

    except Exception as e:
        if conn:
            conn.rollback()
        raise e

    finally:
        # Return connection to pool
        if conn:
            db_pool.putconn(conn)

# Lambda function with connection pooling
connection_pool = None

def lambda_handler(event, context):
    """Lambda with persistent connection pool"""
    global connection_pool

    # Initialize pool once (outside handler reuses)
    if not connection_pool:
        connection_pool = create_connection_pool()

    # Use pooled connection
    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)

    # DynamoDB limits batch to 25 items
    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]

    # Handle unprocessed keys
    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,  # datetime object
        DBInstanceClass='db.t3.medium'
    )

    print(f"Restoring {target_db_id} to {restore_time}")

Quick Reference

Database Service Selection

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

# ❌ NEVER: Create new connection per Lambda invocation
def lambda_handler(event, context):
    conn = psycopg2.connect(
        host='db.example.com',
        database='myapp',
        user='dbadmin',
        password='password'
    )
    # New connection every time = connection exhaustion

# ✅ CORRECT: Reuse connection across invocations
connection = None

def lambda_handler(event, context):
    global connection

    if not connection or connection.closed:
        connection = psycopg2.connect(...)

    # Reuse connection
    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

# ❌ Don't scan entire DynamoDB table
response = table.scan()
items = response['Items']
# Consumes read capacity, expensive, slow

# ✅ Correct: Use query with key condition
response = table.query(
    IndexName='email-index',
    KeyConditionExpression=Key('email').eq(user_email)
)
items = response['Items']
# Efficient, targeted read

❌ 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 databases
• aws-storage.md - S3 for database backups and large objects
• aws-networking.md - VPC endpoints for private database access
• aws-iam-security.md - IAM database authentication and permissions

Last Updated: 2025-10-25 Format Version: 1.0 (Atomic)