AWS DMS for Data Engineers

Full load and CDC, preparing sources, table mappings, landing CDC as Parquet in S3 and merging it into lake tables, tuning, validation and monitoring.

1. What AWS DMS Is

AWS Database Migration Service copies data from a source database to a target, either once (full load), continuously (CDC, change data capture from the transaction log), or both. Data engineers use it to land operational databases in S3 or Redshift and keep them in sync, and to migrate databases to AWS.

2. Preparing the Source for CDC

Without the right logging settings, a CDC task starts but captures nothing or misses columns: check this first when CDC "silently" fails.

3. Table Mappings

{
  "rules": [
    {
      "rule-type": "selection", "rule-id": "1", "rule-name": "include-sales",
      "object-locator": { "schema-name": "SALES", "table-name": "%" },
      "rule-action": "include"
    },
    {
      "rule-type": "selection", "rule-id": "2", "rule-name": "exclude-temp",
      "object-locator": { "schema-name": "SALES", "table-name": "TMP_%" },
      "rule-action": "exclude"
    },
    {
      "rule-type": "transformation", "rule-id": "3", "rule-name": "lowercase-tables",
      "rule-target": "table",
      "object-locator": { "schema-name": "SALES", "table-name": "%" },
      "rule-action": "convert-lowercase"
    },
    {
      "rule-type": "transformation", "rule-id": "4", "rule-name": "drop-pii",
      "rule-target": "column",
      "object-locator": { "schema-name": "SALES", "table-name": "CUSTOMERS", "column-name": "PAN_NUMBER" },
      "rule-action": "remove-column"
    }
  ]
}

4. Landing CDC in S3

aws dms create-endpoint --endpoint-identifier s3-raw-target --endpoint-type target --engine-name s3 \
  --s3-settings '{
    "BucketName": "acme-lake-prod",
    "BucketFolder": "raw/dms",
    "ServiceAccessRoleArn": "arn:aws:iam::111122223333:role/dms-s3-access",
    "DataFormat": "parquet",
    "ParquetVersion": "parquet-2-0",
    "TimestampColumnName": "dms_commit_ts",
    "IncludeOpForFullLoad": true,
    "DatePartitionEnabled": true,
    "DatePartitionSequence": "YYYYMMDD",
    "EncryptionMode": "SSE_KMS",
    "ServerSideEncryptionKmsKeyId": "arn:aws:kms:ap-south-1:111122223333:key/1111-2222"
  }'

aws dms create-replication-task --replication-task-identifier orders-to-s3 \
  --source-endpoint-arn <source-arn> --target-endpoint-arn <target-arn> \
  --replication-instance-arn <instance-arn> \
  --migration-type full-load-and-cdc \
  --table-mappings file://table-mappings.json

Each output row carries an Op column (I insert, U update, D delete) and the commit timestamp. Downstream, keep the latest change per key and MERGE it into an Iceberg, Delta or Hudi table:

from pyspark.sql import functions as F, Window

changes = spark.read.parquet("s3://acme-lake-prod/raw/dms/SALES/ORDERS/")
latest = (changes
    .withColumn("rn", F.row_number().over(
        Window.partitionBy("order_id").orderBy(F.col("dms_commit_ts").desc())))
    .filter("rn = 1").drop("rn"))
latest.createOrReplaceTempView("order_changes")

spark.sql("""
    MERGE INTO glue_catalog.silver.orders t
    USING order_changes s ON t.order_id = s.order_id
    WHEN MATCHED AND s.Op = 'D' THEN DELETE
    WHEN MATCHED THEN UPDATE SET *
    WHEN NOT MATCHED AND s.Op <> 'D' THEN INSERT *
""")

Full-load files have no Op value unless IncludeOpForFullLoad is true; enable it so one MERGE handles both phases.

5. Tuning, Validation & Monitoring

6. Common Interview Questions