Progress
0% View detailed progress dashboard

Cloud Lab: AWS DMS CDC Pipeline

Learn to implement Change Data Capture using AWS Database Migration Service (DMS) to replicate changes from RDS to S3/Redshift. Includes Terraform templates and IAM setup.

Check Prerequisites Start Building

Overview

AWS Database Migration Service (DMS) is a managed service that helps migrate databases to AWS, but it's also a powerful tool for continuous Change Data Capture. Unlike Debezium which requires you to manage Kafka infrastructure, DMS handles replication with minimal operational overhead.

Key differences from Debezium:

  • Fully managed AWS service (no Kafka/Connect cluster to maintain)
  • Native integration with AWS services (S3, Redshift, Kinesis, DynamoDB)
  • Web UI for configuration and monitoring
  • Pay-per-use pricing based on instance hours and data transfer
  • Built-in VPC networking support

Prerequisites

  • AWS Account: Free tier eligible for testing (t3.micro replication instance)
  • AWS CLI: Installed and configured with credentials
  • Terraform: Version 1.0+ (optional, for infrastructure-as-code approach)
  • Source Database: RDS MySQL or PostgreSQL instance with binlog/WAL enabled
  • IAM Permissions: Permissions to create DMS, RDS, S3, and IAM resources

Architecture

We'll build a CDC pipeline that:

  1. Captures changes from an RDS MySQL source database
  2. Uses AWS DMS replication instance to stream changes
  3. Writes CDC events to S3 in Parquet format (for analytics)
  4. Optionally loads data into Redshift for querying
┌─────────────┐      ┌──────────────┐      ┌─────────┐
│ RDS MySQL   │─────▶│  DMS Task    │─────▶│   S3    │
│ (Source)    │      │ (CDC Mode)   │      │ Bucket  │
└─────────────┘      └──────────────┘      └─────────┘
                            │
                            │ (Optional)
                            ▼
                     ┌──────────────┐
                     │  Redshift    │
                     │  (Target)    │
                     └──────────────┘

Lab Setup

Step 1: Prepare Source Database (RDS MySQL)

First, ensure your RDS MySQL instance has binary logging enabled. This is required for DMS to capture changes.

Setting binlog_format=ROW is not sufficient on its own. On RDS MySQL the binary log only exists when automated backups are enabled (backup retention period > 0). With a retention of 0 there is no binlog for DMS to read, regardless of the binlog_format parameter — so enable automated backups on the instance before configuring the CDC task. 

# Check binlog status in MySQL
SHOW VARIABLES LIKE 'log_bin';
-- Should return 'ON'

# Verify binlog format is ROW
SHOW VARIABLES LIKE 'binlog_format';
-- Should return 'ROW'

# Check binlog retention
CALL mysql.rds_show_configuration;
-- binlog retention hours should be > 0 (default: 24 hours)

If binlog is not enabled, update your RDS parameter group:

# Create custom parameter group (AWS CLI)
aws rds create-db-parameter-group \
  --db-parameter-group-name mysql-cdc-params \
  --db-parameter-group-family mysql8.0 \
  --description "MySQL parameters for CDC"

# Enable binlog with ROW format
aws rds modify-db-parameter-group \
  --db-parameter-group-name mysql-cdc-params \
  --parameters "ParameterName=binlog_format,ParameterValue=ROW,ApplyMethod=immediate"

# Apply parameter group to RDS instance
aws rds modify-db-instance \
  --db-instance-identifier my-source-db \
  --db-parameter-group-name mysql-cdc-params \
  --apply-immediately

Step 2: Create IAM Roles

DMS needs IAM roles to access your source database, target S3 bucket, and CloudWatch for logging.

Terraform: IAM Roles (Click to expand) 
# iam.tf
# DMS VPC Management Role
resource "aws_iam_role" "dms_vpc_role" {
  name = "dms-vpc-role"
  
  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [{
      Action = "sts:AssumeRole"
      Effect = "Allow"
      Principal = {
        Service = "dms.amazonaws.com"
      }
    }]
  })
}

resource "aws_iam_role_policy_attachment" "dms_vpc_policy" {
  role       = aws_iam_role.dms_vpc_role.name
  policy_arn = "arn:aws:iam::aws:policy/service-role/AmazonDMSVPCManagementRole"
}

# DMS CloudWatch Logs Role
resource "aws_iam_role" "dms_cloudwatch_role" {
  name = "dms-cloudwatch-logs-role"
  
  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [{
      Action = "sts:AssumeRole"
      Effect = "Allow"
      Principal = {
        Service = "dms.amazonaws.com"
      }
    }]
  })
}

resource "aws_iam_role_policy_attachment" "dms_cloudwatch_policy" {
  role       = aws_iam_role.dms_cloudwatch_role.name
  policy_arn = "arn:aws:iam::aws:policy/service-role/AmazonDMSCloudWatchLogsRole"
}

# S3 Target Role
resource "aws_iam_role" "dms_s3_role" {
  name = "dms-s3-target-role"
  
  assume_role_policy = jsonencode({
    Version = "2012-10-17"
    Statement = [{
      Action = "sts:AssumeRole"
      Effect = "Allow"
      Principal = {
        Service = "dms.amazonaws.com"
      }
    }]
  })
}

resource "aws_iam_role_policy" "dms_s3_policy" {
  name = "dms-s3-access"
  role = aws_iam_role.dms_s3_role.id
  
  policy = jsonencode({
    Version = "2012-10-17"
    Statement = [
      {
        Effect = "Allow"
        Action = [
          "s3:PutObject",
          "s3:DeleteObject",
          "s3:PutObjectTagging"
        ]
        Resource = "arn:aws:s3:::${aws_s3_bucket.cdc_target.bucket}/*"
      },
      {
        Effect = "Allow"
        Action = [
          "s3:ListBucket"
        ]
        Resource = "arn:aws:s3:::${aws_s3_bucket.cdc_target.bucket}"
      }
    ]
  })
}

Step 3: Create S3 Target Bucket

# Create S3 bucket for CDC data
aws s3 mb s3://my-cdc-target-bucket --region us-east-1

# Enable versioning (recommended)
aws s3api put-bucket-versioning \
  --bucket my-cdc-target-bucket \
  --versioning-configuration Status=Enabled
Terraform: S3 Bucket (Click to expand) 
# s3.tf
resource "aws_s3_bucket" "cdc_target" {
  bucket = "my-cdc-target-bucket"
  
  tags = {
    Purpose = "DMS CDC Target"
  }
}

resource "aws_s3_bucket_versioning" "cdc_target" {
  bucket = aws_s3_bucket.cdc_target.id
  
  versioning_configuration {
    status = "Enabled"
  }
}

resource "aws_s3_bucket_public_access_block" "cdc_target" {
  bucket = aws_s3_bucket.cdc_target.id
  
  block_public_acls       = true
  block_public_policy     = true
  ignore_public_acls      = true
  restrict_public_buckets = true
}

Step 4: Create DMS Replication Instance

The replication instance is the compute resource that performs the actual data replication. Use the latest available DMS engine version — the 3.5.x value shown below is illustrative and will age — and note that DMS Serverless is also an option if you'd rather not size and manage a provisioned instance. 

# AWS CLI: Create replication instance (t3.micro for testing)
aws dms create-replication-instance \
  --replication-instance-identifier dms-cdc-instance \
  --replication-instance-class dms.t3.micro \
  --engine-version 3.5.3 \
  --allocated-storage 20 \
  --vpc-security-group-ids sg-xxxxxxxxx \
  --replication-subnet-group-identifier default \
  --no-publicly-accessible \
  --tags Key=Purpose,Value=CDC-Lab

# Check status (takes 5-10 minutes to become available)
aws dms describe-replication-instances \
  --filters Name=replication-instance-id,Values=dms-cdc-instance
Terraform: Replication Instance (Click to expand) 
# dms.tf
resource "aws_dms_replication_subnet_group" "main" {
  replication_subnet_group_id          = "dms-subnet-group"
  replication_subnet_group_description = "DMS replication subnet group"
  subnet_ids                           = var.private_subnet_ids
  
  tags = {
    Name = "dms-subnet-group"
  }
}

resource "aws_dms_replication_instance" "main" {
  replication_instance_id   = "dms-cdc-instance"
  replication_instance_class = "dms.t3.micro"
  engine_version            = "3.5.3"
  allocated_storage         = 20
  
  vpc_security_group_ids = [aws_security_group.dms.id]
  replication_subnet_group_id = aws_dms_replication_subnet_group.main.id
  
  publicly_accessible = false
  multi_az            = false
  
  tags = {
    Name    = "dms-cdc-instance"
    Purpose = "CDC-Lab"
  }
}

Step 5: Create Source and Target Endpoints

Endpoints define connection details for your source database and target S3 bucket.

Source Endpoint (RDS MySQL)

aws dms create-endpoint \
  --endpoint-identifier mysql-source \
  --endpoint-type source \
  --engine-name mysql \
  --server-name my-rds-instance.xxxx.us-east-1.rds.amazonaws.com \
  --port 3306 \
  --username admin \
  --password 'YourSecurePassword' \
  --database-name mydb \
  --ssl-mode require

Target Endpoint (S3)

aws dms create-endpoint \
  --endpoint-identifier s3-target \
  --endpoint-type target \
  --engine-name s3 \
  --s3-settings '{
    "BucketName": "my-cdc-target-bucket",
    "BucketFolder": "cdc-data",
    "CompressionType": "GZIP",
    "DataFormat": "parquet",
    "ServiceAccessRoleArn": "arn:aws:iam::ACCOUNT:role/dms-s3-target-role",
    "TimestampColumnName": "cdc_timestamp",
    "ParquetVersion": "parquet-2-0",
    "EnableStatistics": true
  }'
Terraform: Endpoints (Click to expand) 
# dms-endpoints.tf
resource "aws_dms_endpoint" "source" {
  endpoint_id   = "mysql-source"
  endpoint_type = "source"
  engine_name   = "mysql"
  
  server_name = aws_db_instance.source.address
  port        = 3306
  username    = var.db_username
  password    = var.db_password
  database_name = var.db_name
  
  ssl_mode = "require"
  
  tags = {
    Name = "mysql-source-endpoint"
  }
}

resource "aws_dms_endpoint" "target" {
  endpoint_id   = "s3-target"
  endpoint_type = "target"
  engine_name   = "s3"
  
  s3_settings {
    bucket_name              = aws_s3_bucket.cdc_target.bucket
    bucket_folder            = "cdc-data"
    compression_type         = "GZIP"
    data_format              = "parquet"
    service_access_role_arn  = aws_iam_role.dms_s3_role.arn
    timestamp_column_name    = "cdc_timestamp"
    parquet_version          = "parquet-2-0"
    enable_statistics        = true
  }
  
  tags = {
    Name = "s3-target-endpoint"
  }
}

Step 6: Test Endpoints

# Test source endpoint
aws dms test-connection \
  --replication-instance-arn arn:aws:dms:region:account:rep:xxx \
  --endpoint-arn arn:aws:dms:region:account:endpoint:xxx

# Test target endpoint
aws dms test-connection \
  --replication-instance-arn arn:aws:dms:region:account:rep:xxx \
  --endpoint-arn arn:aws:dms:region:account:endpoint:yyy

# Check test results
aws dms describe-connections \
  --filters Name=replication-instance-arn,Values=arn:aws:dms:xxx

Step 7: Create DMS Replication Task

The replication task defines what data to replicate and how. We'll use full-load-and-cdc mode: DMS first records the current source log position (a start LSN/SCN), takes a full snapshot of the selected tables, then resumes streaming changes from that recorded position — so no committed change is missed in the handoff between the snapshot and ongoing CDC.

Create a table mappings file (table-mappings.json):

{
  "rules": [
    {
      "rule-type": "selection",
      "rule-id": "1",
      "rule-name": "include-all-tables",
      "object-locator": {
        "schema-name": "mydb",
        "table-name": "%"
      },
      "rule-action": "include"
    },
    {
      "rule-type": "transformation",
      "rule-id": "2",
      "rule-name": "add-cdc-metadata",
      "rule-target": "column",
      "object-locator": {
        "schema-name": "%",
        "table-name": "%"
      },
      "rule-action": "add-column",
      "value": "cdc_operation",
      "expression": "$operation",
      "data-type": {
        "type": "string",
        "length": 10
      }
    }
  ]
}

Create the task:

aws dms create-replication-task \
  --replication-task-identifier cdc-task-mysql-to-s3 \
  --source-endpoint-arn arn:aws:dms:region:account:endpoint:source-id \
  --target-endpoint-arn arn:aws:dms:region:account:endpoint:target-id \
  --replication-instance-arn arn:aws:dms:region:account:rep:instance-id \
  --migration-type full-load-and-cdc \
  --table-mappings file://table-mappings.json \
  --replication-task-settings '{
    "Logging": {
      "EnableLogging": true,
      "LogComponents": [
        {"Id": "SOURCE_CAPTURE", "Severity": "LOGGER_SEVERITY_INFO"},
        {"Id": "TARGET_LOAD", "Severity": "LOGGER_SEVERITY_INFO"}
      ]
    },
    "ChangeProcessingTuning": {
      "BatchApplyEnabled": true,
      "BatchApplyTimeoutMin": 1,
      "BatchApplyTimeoutMax": 30,
      "MinTransactionSize": 1000,
      "CommitTimeout": 1
    }
  }'
Terraform: Replication Task (Click to expand) 
# dms-task.tf
resource "aws_dms_replication_task" "main" {
  replication_task_id       = "cdc-task-mysql-to-s3"
  migration_type            = "full-load-and-cdc"
  replication_instance_arn  = aws_dms_replication_instance.main.replication_instance_arn
  source_endpoint_arn       = aws_dms_endpoint.source.endpoint_arn
  target_endpoint_arn       = aws_dms_endpoint.target.endpoint_arn
  
  table_mappings = jsonencode({
    rules = [
      {
        rule-type = "selection"
        rule-id   = "1"
        rule-name = "include-all-tables"
        object-locator = {
          schema-name = "mydb"
          table-name  = "%"
        }
        rule-action = "include"
      }
    ]
  })
  
  replication_task_settings = jsonencode({
    Logging = {
      EnableLogging = true
      LogComponents = [
        {Id = "SOURCE_CAPTURE", Severity = "LOGGER_SEVERITY_INFO"},
        {Id = "TARGET_LOAD", Severity = "LOGGER_SEVERITY_INFO"}
      ]
    }
    ChangeProcessingTuning = {
      BatchApplyEnabled    = true
      BatchApplyTimeoutMin = 1
      BatchApplyTimeoutMax = 30
      MinTransactionSize   = 1000
      CommitTimeout        = 1
    }
  })
  
  tags = {
    Name = "cdc-task-mysql-to-s3"
  }
}

Step 8: Start the Replication Task

# Start the CDC task
aws dms start-replication-task \
  --replication-task-arn arn:aws:dms:region:account:task:task-id \
  --start-replication-task-type start-replication

# Monitor task progress
aws dms describe-replication-tasks \
  --filters Name=replication-task-arn,Values=arn:aws:dms:xxx

# Watch CloudWatch Logs for detailed progress
aws logs tail /aws/dms/tasks/cdc-task-mysql-to-s3 --follow

Verification

⚠️ DMS delivery is at-least-once — dedupe downstream

DMS CDC delivery to S3 (and onward to Redshift) is at-least-once, not exactly-once. On a task restart or resume, DMS re-reads from the last checkpointed log position and can re-emit changes it already wrote, producing duplicate rows in S3/Redshift. This is expected behavior, not a bug.

Make the consumer idempotent: upsert on the primary key, ordered by the change's log position (the source binlog coordinate / LSN / SCN), not by cdc_timestamp. The timestamp is coarse and can tie or move backwards across restarts; the log position is the only monotonic ordering of committed changes. End-to-end exactly-once across DMS and your warehouse is not achievable — correctness comes from an idempotent sink, not from the pipe.

1. Insert Test Data

-- Connect to source RDS MySQL
mysql -h my-rds-instance.xxxx.rds.amazonaws.com -u admin -p

-- Create test table and insert data
USE mydb;

CREATE TABLE customers (
  id INT PRIMARY KEY AUTO_INCREMENT,
  name VARCHAR(100),
  email VARCHAR(100),
  created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

INSERT INTO customers (name, email) VALUES 
  ('Alice Johnson', 'alice@example.com'),
  ('Bob Smith', 'bob@example.com'),
  ('Carol White', 'carol@example.com');

-- Update a record
UPDATE customers SET email = 'alice.j@example.com' WHERE id = 1;

-- Delete a record
DELETE FROM customers WHERE id = 3;

2. Check S3 for CDC Files

# List CDC files in S3
aws s3 ls s3://my-cdc-target-bucket/cdc-data/mydb/customers/ --recursive

# Download and inspect a Parquet file
aws s3 cp s3://my-cdc-target-bucket/cdc-data/mydb/customers/LOAD00000001.parquet .

# Use parquet-tools to read (install: pip install parquet-tools)
parquet-tools show LOAD00000001.parquet

3. Monitor DMS Metrics

Key CloudWatch metrics to monitor:

  • CDCLatencySource: Time lag between source change and capture (should be < 10s)
  • CDCLatencyTarget: Time lag between capture and target write
  • FullLoadThroughputRowsSource: Rows per second being read
  • FullLoadThroughputRowsTarget: Rows per second being written
# Query CDC lag metric
aws cloudwatch get-metric-statistics \
  --namespace AWS/DMS \
  --metric-name CDCLatencySource \
  --dimensions Name=ReplicationTaskIdentifier,Value=cdc-task-mysql-to-s3 \
  --start-time 2024-01-01T00:00:00Z \
  --end-time 2024-01-01T23:59:59Z \
  --period 300 \
  --statistics Average

Vendor-Specific Gotchas

⚠️ Common Issues and Solutions

1. Binlog Retention Too Short

Issue: DMS task fails with "binlog position no longer available" after replication instance restart.

Solution: Increase binlog retention on RDS: 

CALL mysql.rds_set_configuration('binlog retention hours', 168); -- 7 days

2. IAM Role Trust Relationship

Issue: "Access Denied" when DMS tries to write to S3.

Solution: Ensure IAM role has correct trust relationship with dms.amazonaws.com. Check assume role policy: 

{
  "Version": "2012-10-17",
  "Statement": [{
    "Effect": "Allow",
    "Principal": {
      "Service": "dms.amazonaws.com"
    },
    "Action": "sts:AssumeRole"
  }]
}

3. VPC Connectivity

Issue: DMS cannot connect to RDS even with correct credentials.

Solution: Verify security group rules allow traffic from DMS replication instance security group to RDS on port 3306. Use VPC Flow Logs to debug.

4. LOB (Large Object) Handling

Issue: Tasks stall when replicating tables with BLOB/TEXT columns.

Solution: Adjust LOB settings in task configuration: 

"TargetMetadata": {
  "SupportLobs": true,
  "FullLobMode": false,
  "LobChunkSize": 64,
  "LimitedSizeLobMode": true,
  "LobMaxSize": 32
}

5. S3 Parquet File Organization

DMS creates files with timestamps in UTC. Use partitioning to organize by date for efficient querying: 

"BucketFolder": "cdc-data",
"DatePartitionEnabled": true,
"DatePartitionSequence": "YYYYMMDD",
"DatePartitionDelimiter": "/"

Results in: s3://bucket/cdc-data/mydb/customers/20240615/LOAD00001.parquet

Cost Considerations

AWS DMS pricing includes the components below. These figures are approximate and vary by region and over time — check the current DMS pricing page before estimating costs.

  • Replication Instance: ~$0.036/hour for dms.t3.micro (~$26/month)
  • S3 Storage: $0.023/GB per month (Standard tier)
  • Data Transfer: Free within same region; $0.09/GB for cross-region
  • CloudWatch Logs: $0.50/GB ingested

Tip: Use t3.micro for development/testing. For production, choose instance size based on expected change volume and latency requirements.

Next Steps

  • Query CDC Data: Load Parquet files into Athena or Redshift Spectrum for SQL analysis
  • Set Up Alerts: Create CloudWatch alarms for CDCLatency > threshold
  • Implement Retry Logic: Handle transient failures with task restart automation
  • Add More Tables: Expand table mappings to include additional schemas
  • Optimize Performance: Tune BatchApplyEnabled and CommitTimeout for your workload

DMS vs. Debezium: When to Choose Each

Aspect AWS DMS Debezium
Infrastructure Fully managed, no ops overhead Self-managed Kafka + Connect cluster
Target Flexibility AWS services (S3, Redshift, Kinesis) Kafka topics (any downstream consumer)
Event Format Proprietary or Parquet/CSV Rich JSON with before/after, metadata
Customization Limited (table mappings, filters) High (SMTs, custom connectors)
Cost Pay-per-instance-hour (~$26-500+/mo) Infrastructure costs (EC2, storage)
Best For AWS-native stacks, simple pipelines Event-driven architectures, Kafka ecosystems

Additional Resources

View All Cloud Labs