PolarDB:Data migration

How it works

The migration uses three tools across five stages:

Stage 1 — Full synchronization (nimo-shake): The tool auto-creates table structures in the target PolarDB cluster that mirror the source, then reads all data concurrently via DynamoDB Scan operations and writes it in batches using BatchWriteItem.

Stage 2 — Incremental synchronization (nimo-shake): After full synchronization completes, the tool switches automatically to capture ongoing changes (inserts, deletes, and updates) via AWS DynamoDB Streams and applies them to PolarDB. This phase supports resumable transfers.

Stage 3 — Consistency check (nimo-full-check): Run at any point during or after synchronization. The tool reads both source and target concurrently, compares records by primary key, and produces a detailed diff report.

Stage 4 — Reverse synchronization (PolarDBBackSync, optional): Before cutover, set up a reverse link from PolarDB back to DynamoDB. If cutover causes issues, this link lets data flow back to the source as a rollback safeguard. PolarDBBackSync uses Flink to capture real-time changes from PolarDB and applies them to DynamoDB using PutItem or DeleteItem operations based on the change type. It is built on Realtime Compute for Apache Flink.

Stage 5 — Business cutover: When incremental synchronization lag is near-zero and the consistency check shows no differences, stop writes to the source, confirm the last sync, then switch your application connections to PolarDB.

Usage notes

• Schedule during off-peak hours: Assess database capacity and run the migration when load is lowest to minimize performance impact.

• Restrict write access before cutover: Grant write permissions to the target PolarDB cluster only for the account used by the synchronization tool. This prevents accidental data contamination before the official switch.

Prerequisites

Before you begin, ensure that you have:

• Migration tools — Download NimoShake.tar.gz. This archive contains both nimo-shake and nimo-full-check

• (Optional) Reverse migration tool — Download PolarDBBackSync.jar if you plan to configure a rollback path

• A PolarDB for PostgreSQL cluster — Enable DynamoDB compatibility, obtain the DynamoDB endpoint, and create a dedicated DynamoDB account to get the AccessKey credentials for API access

• (Optional) `wal_level` set to `logical` — Required only if you plan to configure reverse synchronization. This change requires a cluster restart, so complete it before starting the migration

• AWS DynamoDB credentials — AccessKey ID and Secret Access Key for the source DynamoDB

• A runtime host — An ECS instance or any server that can reach both the PolarDB cluster and AWS DynamoDB

Migrate data

SELECT * FROM pg_replication_slots;

Appendix: Simulate real-time traffic for testing

To validate the migration process in a test environment with continuous data writes, use the following Go sample code. It creates a DynamoDB table, inserts 100 initial items, then updates one record per second to simulate business traffic.

This code is for testing only. Do not run it during a production migration.

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)
}