数据迁移指南

迁移流程概述

迁移过程主要分为五个阶段，由nimo-shake（数据同步，包括全量同步与增量同步）、nimo-full-check（数据校验）和PolarDBBackSync（数据反向同步）三个核心工具协同完成。

1. 全量同步（Full Synchronization）

   • 工具：nimo-shake

   • 过程：工具首先自动在目标PolarDB集群中创建与源端一致的表结构。随后，通过并发Scan操作高效读取源端数据库的全量数据，并使用BatchWriteItem批量写入目标集群。

2. 增量同步（Incremental Synchronization）

   • 工具：nimo-shake

   • 过程：全量同步完成后，工具会自动利用AWS DynamoDB Streams机制，实时捕获源端自迁移启动以来的所有数据变更（增、删、改），并将其同步到目标集群，确保数据最终一致。该过程支持断点续传。

3. 一致性校验（Consistency Validation）

   • 工具：nimo-full-check

   • 过程：在数据同步期间或之后，可随时运行此工具。它会并发地从源端和目标端读取数据，按主键进行比对，并生成详细的差异报告，以验证数据完整性。

4. （可选）反向同步（Reverse Synchronization）

   • 工具：PolarDBBackSync.jar (基于阿里云实时计算Flink版)

   • 过程：验证完数据一致性后，为确保业务回滚时数据的完整性，可以创建从PolarDB PostgreSQL版到源端DynamoDB的反向同步。该工具基于Flink实时捕获源端PolarDB的变更数据，并根据变更类型调用DynamoDB的PutItem或DeleteItem接口同步更新DynamoDB的数据。

5. 业务割接（Business Cutover）

   • 过程：当增量数据延迟极低且一致性校验无差异后，短暂停止业务写入，待所有数据同步完毕，即可将应用连接切换至PolarDB集群，完成迁移。

注意事项

• 性能影响：数据迁移过程，尤其是全量同步阶段，会对源数据库和目标数据库产生一定的读写负载。建议您在业务低峰期执行迁移，并提前评估数据库的承载能力。

• 安全配置：在业务割接前，建议对目标PolarDB集群的写入权限进行管控，仅允许数据同步工具的账号写入，防止意外数据污染。

准备工作

在开始迁移前，请确保您已完成以下准备工作：

1. 获取工具包：

   1. 迁移工具包：NimoShake.tar.gz。其中包含nimo-shake和nimo-full-check两种迁移工具包。

   2. （可选）反向迁移工具包：PolarDBBackSync.jar。

2. PolarDB集群：

   1. 为已有集群或新集群开启兼容DynamoDB能力，并获取DynamoDB访问地址和创建DynamoDB专用账号用于API访问的身份凭证（AccessKey）。

   2. （可选）参数配置：若需配置反向同步，则需将PolarDB集群的wal_level参数修改为logical。由于该参数的调整需重启集群，建议在整体迁移流程开始之前完成此项设置。

3. AWS DynamoDB：

   1. 获取AWS DynamoDB的访问凭证（AccessKey ID和Secret Access Key）。

   2. 为需要迁移的源端DynamoDB表开启DynamoDB Streams功能。

4. 运行环境：准备一台ECS实例或其他能够与PolarDB集群及AWS DynamoDB连接的服务器，以便运行迁移工具包。

迁移实施步骤

附录：为测试环境模拟实时业务流量

如果您希望在测试环境中模拟一个真实的、持续有数据写入的迁移场景，可以使用以下Go语言示例代码。该代码会向源端DynamoDB表周期性地写入和更新数据。

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