PolarDB:データ移行

移行の概要

移行プロセスは主に 5 段階に分けられ、以下の 3 つのコアツールが連携して実行されます：nimo-shake（データ同期：フル同期および増分同期を含む）、nimo-full-check（データ検証）、および PolarDBBackSync（逆同期）。

1. フル同期

   • ツール：nimo-shake

   • 処理内容：ツールはまず、ソース側と同一のテーブル構造をターゲットの PolarDB クラスター内に自動的に作成します。その後、並列実行される Scan 操作によりソースデータベースから全データを効率よく読み込み、BatchWriteItem を使用してバッチ単位でターゲットクラスターへ書き込みます。

2. 増分同期

   • ツール：nimo-shake

   • 処理内容：フル同期完了後、ツールは AWS DynamoDB Streams を自動的に活用し、移行開始以降のソース側におけるすべてのデータ変更（挿入、削除、更新）をキャプチャします。これらの変更はターゲットクラスターへ適用され、最終的なデータ整合性を保証します。このプロセスは再開可能な転送をサポートします。

3. 整合性チェック

   • ツール：nimo-full-check

   • 処理内容：データ同期中または同期後にいつでもこのツールを実行できます。ツールはソースおよびターゲットから並列にデータを読み取り、プライマリキーに基づいて比較を行い、データ整合性を検証するための詳細な差分レポートを生成します。

4. （任意）逆同期

   • ツール：PolarDBBackSync.jar（Realtime Compute for Apache Flink を基盤とする）

   • 処理内容：データ整合性の検証が完了した後、ロールバック時のデータ整合性を確保するために、PolarDB for PostgreSQL からソースの DynamoDB へ向けて逆同期タスクを設定できます。このツールは Flink を使用してソースの PolarDB クラスターからリアルタイムの変更をキャプチャし、変更タイプに応じて PutItem または DeleteItem 操作を用いて DynamoDB へ適用します。

5. ビジネスカットオーバー

   • 処理内容：増分データの遅延がほぼゼロとなり、整合性チェックで差分が検出されない状態になった時点で、アプリケーションによる書き込みを一時停止します。残りのデータの同期が完了した後、アプリケーション接続先を 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）を取得してください。

4. 実行環境：移行ツールを実行するため、PolarDB クラスターおよび AWS DynamoDB の両方に接続可能な ECS インスタンスまたはその他のサーバーを準備してください。

操作手順

付録：リアルタイムトラフィックのシミュレーション

テスト環境で継続的なデータ書き込みを伴う現実的な移行シナリオをシミュレートしたい場合、以下の 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"
)

// --- ソースの AWS DynamoDB 用構成 ---
var (
    region    = "cn-north-1"       // AWS DynamoDB リージョン
    accessKey = "your-aws-access-key" // AWS DynamoDB アクセスキー
    secretKey = "your-aws-secret-key" // AWS DynamoDB シークレットキー
)

// --- DynamoDB クライアント作成用ヘルパー関数 ---
func createClient() (*dynamodb.Client, context.Context) {
    ctx := context.Background()
    sdkConfig, err := config.LoadDefaultConfig(ctx, config.WithRegion(region))
    if err != nil {
        log.Fatalf("AWS 構成の読み込みに失敗しました: %v", err)
    }
    client := dynamodb.NewFromConfig(sdkConfig, func(o *dynamodb.Options) {
        o.Credentials = credentials.NewStaticCredentialsProvider(accessKey, secretKey, "")
    })
    return client, ctx
}

// --- テーブルの作成および初期データの投入用関数 ---
func initializeData(client *dynamodb.Client, ctx context.Context) {
    tableName := "src1" // 例：テーブル名

    // テーブルが存在しない場合に作成
    _, 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 {
		// テーブルが既に存在する場合は無視し、その他のエラーの場合は失敗
        if _, ok := err.(*types.ResourceInUseException); !ok {
			log.Fatalf("%s に対する CreateTable が失敗しました: %v", tableName, err)
		}
    }

    fmt.Printf("テーブル '%s' がアクティブになるまで待機中...\n", tableName)
    waiter := dynamodb.NewTableExistsWaiter(client)
    err = waiter.Wait(ctx, &dynamodb.DescribeTableInput{TableName: &tableName}, 5*time.Minute)
    if err != nil {
        log.Fatalf("テーブル %s の待機に失敗しました: %v", tableName, err)
    }

    // 100 件のサンプルアイテムを挿入
    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("'%s' に 100 件の初期アイテムを挿入しました。\n", tableName)
}

// --- 連続的なビジネトラフィックのシミュレーション用関数 ---
func simulateTraffic(client *dynamodb.Client, ctx context.Context) {
    tableName := "src1"
    fmt.Println("トラフィックのシミュレーションを開始します（Ctrl+C で停止）。")
    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("更新エラー: %v\n", err)
        } else {
            fmt.Printf("pk=%s を新しい val=%s で更新しました\n", pk, newValue)
        }
        i++
        time.Sleep(1 * time.Second) // 1 秒ごとに 1 件のレコードを更新
    }
}

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