PolarDB:Migrate from Amazon DynamoDB to PolarDB for PostgreSQL

Migration overview

The migration process has five main stages and uses three command-line tools:  nimo-shake for the full and incremental synchronization, nimo-full-check for validation, and PolarDBBackSync for reverse synchronization.

1. Full synchronization

   • Tool: nimo-shake

   • Process: The nimo-shake tool first connects to your source DynamoDB table and automatically creates a matching table structure in your target PolarDB cluster. It then reads all data from the source table in parallel and writes it to the target cluster in batches.

2. Incremental synchronization

   • Tool: nimo-shake

   • Process: After the full copy is complete, nimo-shake automatically begins capturing data changes (inserts, updates, and deletes) from the source database in real time. It uses Amazon DynamoDB Streams to achieve this and applies the changes to the target cluster, ensuring the two databases remain in sync. This process supports resumable transfers if interrupted.

3. Consistency check

   • Tool: nimo-full-check

   • Process: At any time during or after the synchronization, you can run the nimo-full-check tool. It reads data from both the source and target databases, compares records by their primary key, and generates a detailed report of any differences.

4. (Optional) Reverse synchronization

   • Tool: PolarDBBackSync.jar (Based on Realtime Compute for Apache Flink)

   • Process: After the consistency check, you can create a reverse synchronization task from PolarDB for PostgreSQL to DynamoDB for data integrity. This tool is implemented based on Realtime Compute for Apache Flink to capture data changes in the source PolarDB cluster and call the PutItem or DeleteItem operation to update data in the DynamoDB based on the type of the changed data.

5. Business cutover

   • Process: Once the incremental data lag is near zero and the validation tool reports no data differences, you can perform the cutover. This involves a brief service pause to stop writes to the source database, followed by redirecting your application's connection to the PolarDB cluster.

Notes

• Performance impact: The data migration process, especially the full synchronization phase, creates I/O load on the source and target databases. We recommend that you run the migration during off-peak hours and evaluate your database capacity beforehand.

• Security configuration: Before the business cutover, manage the write permissions for the target PolarDB cluster. To prevent accidental data corruption, grant write permissions only to the account used by the data synchronization tool.

Preparations

Before you start the migration, complete the following preparations:

1. Obtain the tools.

   1. Download th tools for data synchronization: NimoShake.tar.gz. The package includes the nimo-shake and nimo-full-check tools.

   2. (Optional) Download the tool for reverse synchronization: PolarDBBackSync.jar.

2. Configure the PolarDB cluster.

   1. Enable the DynamoDB-compatible feature for the cluster. Then, obtain the DynamoDB endpoint of the cluster and create a dedicated DynamoDB account for API access.

   2. (Optional) Configure parameters: If you want to configure reverse synchronization, set the wal_level parameter of the PolarDB cluster to logical. The change of this parameter takes effect only after the cluster is restart. We recommend changing this parameter before the synchronization.

3. Configure the AWS DynamoDB instance.

   1. Get the access credential (AccessKey ID and Secret Access Key) for the AWS DynamoDB instance.

   2. Enable the DynamoDB Streams feature for the source DynamoDB tables.

4. Prepare the runtime environment.

   Create an ECS instance or a server that can connect to the PolarDB cluster and the AWS DynamoDB instance.

Migration procedure

Step 1: Configure and start data synchronization

1. Unzip the nimo-shake tool package and edit the conf/nimo-shake.conf configuration file in the NimoShake/nimo-shake directory. The following table describes the core configuration items:

   Parameter	Description	Example value
   sync_mode	Synchronization mode. all specifies full and incremental synchronization. full specifies only full synchronization.	all
   source.access_key_id	The AccessKey ID of the source AWS DynamoDB instance.	AKIAIOSFODNN7...
   source.secret_access_key	The Secret Access Key of the source AWS DynamoDB instance.	wJalrXUtnFEMI...
   source.region	The region where the source AWS DynamoDB instance resides.	cn-north-1
   source.endpoint_url	The source endpoint. For AWS DynamoDB, leave this item empty.	-
   target.address	The DynamoDB endpoint of the target PolarDB. Include http:// and the port number.	http://pe-xxx.rwlb.rds...
   target.access_key_id	The AccessKey of the DynamoDB account for the target PolarDB.	your-polardb-access-key
   target.secret_access_key	The SecretKey of the DynamoDB account for the target PolarDB.	your-polardb-secret-key
   filter.collection.white	Whitelist for table filtering. Only the specified tables are synchronized. Separate multiple table names with a semicolon (;). Note Do not use the whitelist and blacklist at the same time. If you specify both, all tables are synchronized.	c1;c2
   filter.collection.black	Blacklist for table filtering. The specified tables are excluded from synchronization. Separate multiple table names with a semicolon (;). Note Do not use the whitelist and blacklist at the same time. If you specify both, all tables are synchronized.	c1;c2

2. Select the binary file that corresponds to your operating system architecture and start the sync task.

   Note

   Run the task with the nohup option to prevent the sync task from being interrupted if the terminal session disconnects.

   # Return to the nimo-shake directory
   cd ..
   # For example, run the task on a Linux server with the x86-64 architecture
   nohup ./bin/nimo-shake.linux.amd64 -conf=./conf/nimo-shake.conf > nimo-shake.log 2>&1 &

   The program first performs a full synchronization. After it is complete, the program automatically switches to the incremental synchronization phase and continues to run.

Appendix: Simulate real-time service traffic for a test environment

To simulate a realistic migration scenario with continuous data writes in a test environment, you can use the following Go sample code. The code periodically writes and updates data in the source DynamoDB table.

package main

import (
	"context"
	"fmt"
	"log"
	"math/rand"
	"time"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/config"
	"github.com/aws/aws-sdk-go-v2/credentials"
	"github.com/aws/aws-sdk-go-v2/service/dynamodb"
	"github.com/aws/aws-sdk-go-v2/service/dynamodb/types"
)

// --- Configuration for your source AWS DynamoDB ---
var (
    region    = "cn-north-1"       // AWS DynamoDB region
    accessKey = "your-aws-access-key" // AWS DynamoDB access key
    secretKey = "your-aws-secret-key" // AWS DynamoDB secret key
)

// --- Helper function to create a DynamoDB client ---
func createClient() (*dynamodb.Client, context.Context) {
    ctx := context.Background()
    sdkConfig, err := config.LoadDefaultConfig(ctx, config.WithRegion(region))
    if err != nil {
        log.Fatalf("Failed to load AWS config: %v", err)
    }
    client := dynamodb.NewFromConfig(sdkConfig, func(o *dynamodb.Options) {
        o.Credentials = credentials.NewStaticCredentialsProvider(accessKey, secretKey, "")
    })
    return client, ctx
}

// --- Function to create a table and populate it with initial data ---
func initializeData(client *dynamodb.Client, ctx context.Context) {
    tableName := "src1" // Example table name

    // Create table if not exists
    _, err := client.CreateTable(ctx, &dynamodb.CreateTableInput{
        TableName: &tableName,
        AttributeDefinitions: []types.AttributeDefinition{
            {AttributeName: aws.String("pk"), AttributeType: types.ScalarAttributeTypeS},
        },
        KeySchema: []types.KeySchemaElement{
            {AttributeName: aws.String("pk"), KeyType: types.KeyTypeHash},
        },
        ProvisionedThroughput: &types.ProvisionedThroughput{
            ReadCapacityUnits:  aws.Int64(100),
            WriteCapacityUnits: aws.Int64(100),
        },
    })
    if err != nil {
		// Ignore if table already exists, fail on other errors
        if _, ok := err.(*types.ResourceInUseException); !ok {
			log.Fatalf("CreateTable failed for %s: %v", tableName, err)
		}
    }

    fmt.Printf("Waiting for table '%s' to become active...\n", tableName)
    waiter := dynamodb.NewTableExistsWaiter(client)
    err = waiter.Wait(ctx, &dynamodb.DescribeTableInput{TableName: &tableName}, 5*time.Minute)
    if err != nil {
        log.Fatalf("Waiter failed for table %s: %v", tableName, err)
    }

    // Insert 100 sample items
    for i := 0; i < 100; i++ {
        pk := fmt.Sprintf("%s_user_%03d", tableName, i)
        item := map[string]types.AttributeValue{
            "pk":  &types.AttributeValueMemberS{Value: pk},
            "val": &types.AttributeValueMemberN{Value: fmt.Sprintf("%d", i)},
        }
        client.PutItem(ctx, &dynamodb.PutItemInput{TableName: &tableName, Item: item})
    }
    fmt.Printf("Inserted 100 initial items into '%s'.\n", tableName)
}

// --- Function to simulate continuous business traffic ---
func simulateTraffic(client *dynamodb.Client, ctx context.Context) {
    tableName := "src1"
    fmt.Println("Starting periodic updates to simulate traffic. Press Ctrl+C to stop.")
    i := 0
    for {
        pk := fmt.Sprintf("%s_user_%03d", tableName, i%100)
        newValue := fmt.Sprintf("%d", rand.Intn(1000))
        _, err := client.UpdateItem(ctx, &dynamodb.UpdateItemInput{
            TableName: &tableName,
            Key: map[string]types.AttributeValue{
                "pk": &types.AttributeValueMemberS{Value: pk},
            },
            ExpressionAttributeValues: map[string]types.AttributeValue{
                ":newval": &types.AttributeValueMemberN{Value: newValue},
            },
            UpdateExpression: aws.String("SET val = :newval"),
        })
        if err != nil {
            fmt.Printf("Update error: %v\n", err)
        } else {
            fmt.Printf("Updated pk=%s with new val=%s\n", pk, newValue)
        }
        i++
        time.Sleep(1 * time.Second) // Update one record per second
    }
}

func main() {
    client, ctx := createClient()
    initializeData(client, ctx)
    simulateTraffic(client, ctx)
}