PolarDB：資料移轉指南

遷移流程概述

遷移過程主要分為五個階段，由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 (基於阿里雲Realtime ComputeFlink版)

   • 過程：驗證完資料一致性後，為確保業務復原時資料的完整性，可以建立從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)
}