使用遷移工具將Amazon DynamoDB資料移轉至PolarDB PostgreSQL版-雲原生資料庫 PolarDB-阿里雲 - PolarDB

本指南為您提供了從Amazon DynamoDB遷移至PolarDB DynamoDB相容版詳盡的操作步驟和最佳實務。PolarDB提供了一套專用的遷移工具，通過全量同步與增量同步處理相結合的方式，協助您實現平滑、低停機時間的資料移轉。

遷移流程概述

遷移過程主要分為五個階段，由nimo-shake（資料同步，包括全量同步與增量同步處理）、nimo-full-check（資料校正）和PolarDBBackSync（資料反向同步）三個核心工具協同完成。

全量同步（Full Synchronization）
- 工具：nimo-shake
- 過程：工具首先自動在目標PolarDB叢集中建立與源端一致的表結構。隨後，通過並發Scan操作高效讀取源端資料庫的全量資料，並使用BatchWriteItem批量寫入目的地組群（預設通過PostgreSQL協議直寫，詳見步驟一中的全量同步PostgreSQL直寫入模式）。
增量同步處理（Incremental Synchronization）
- 工具：nimo-shake
- 過程：全量同步完成後，工具會自動利用AWS DynamoDB Streams機制，即時捕獲源端自遷移啟動以來的所有資料變更（增、刪、改），並將其同步到目的地組群，確保資料最終一致。該過程支援斷點續傳。
一致性校正（Consistency Validation）
- 工具：nimo-full-check
- 過程：在資料同步期間或之後，可隨時運行此工具。它會並發地從源端和目標端讀取資料，按主鍵進行比對，並產生詳細的差異報告，以驗證資料完整性。
（可選）反向同步（Reverse Synchronization）
- 工具：PolarDBBackSync.jar (基於阿里雲Realtime ComputeFlink版)
- 過程：驗證完資料一致性後，為確保業務復原時資料的完整性，可以建立從PolarDB PostgreSQL版到源端DynamoDB的反向同步。該工具基於Flink即時捕獲源端PolarDB的變更資料，並根據變更類型調用DynamoDB的PutItem或DeleteItem介面同步更新DynamoDB的資料。
業務割接（Business Cutover）
- 過程：當增量資料延遲極低且一致性校正無差異後，短暫停止業務寫入，待所有資料同步完畢，即可將應用串連切換至PolarDB叢集，完成遷移。

注意事項

效能影響：資料移轉過程，尤其是全量同步階段，會對來源資料庫和目標資料庫產生一定的讀寫負載。建議您在業務低峰期執行遷移，並提前評估資料庫的承載能力。
安全配置：在業務割接前，建議對目標PolarDB叢集的寫入許可權進行管控，僅允許資料同步工具的帳號寫入，防止意外資料汙染。

準備工作

在開始遷移前，請確保您已完成以下準備工作：

擷取工具包：
1. 遷移工具包：NimoShake-20260807.zip。其中包含nimo-shake，nimo-full-check和nimo-repair三種遷移工具包。
2. （可選）反向遷移工具包：PolarDBBackSync.jar。
PolarDB叢集：
1. 為已有叢集或新叢集開啟相容DynamoDB能力，並擷取DynamoDB訪問地址和建立DynamoDB專用帳號用於API訪問的身份憑證（AccessKey）。
2. （可選）參數配置：若需配置反向同步，則需將PolarDB叢集的wal_level參數修改為logical。由於該參數的調整需重啟叢集，建議在整體遷移流程開始之前完成此項設定。
AWS DynamoDB：
1. 擷取AWS DynamoDB的訪問憑證（AccessKey ID和Secret Access Key）。
運行環境：準備一台ECS執行個體或其他能夠與PolarDB叢集及AWS DynamoDB已連線的服務器，以便運行遷移工具包。

遷移實施步驟

步驟一：配置並啟動資料同步

解壓 NimoShake.zip，進入NimoShake目錄，編輯統一設定檔conf/nimo.conf。nimo-shake、nimo-full-check、nimo-repair三個工具共用此檔案，每個工具唯讀取自己能識別的參數。以下是啟動同步所需的核心配置項：

參數	說明	樣本值
`sync_mode`	同步模式。`all`表示全量+增量，`full`表示僅全量。	`all`
`source.access_key_id`	源端AWS DynamoDB的AccessKey ID。	`AKIAIOSFODNN7...`
`source.secret_access_key`	源端AWS DynamoDB的Secret Access Key。	`wJalrXUtnFEMI...`
`source.region`	源端AWS DynamoDB所在的地區。	`cn-north-1`
`target.endpoint_url`	目標PolarDB的DynamoDB訪問地址（含連接埠）。	`http://pe-xxx.rwlb.rds...`
`target.access_key_id`	目標PolarDB的DynamoDB帳號AccessKey。	`your-polardb-access-key`
`target.secret_access_key`	目標PolarDB的DynamoDB帳號SecretKey。	`your-polardb-secret-key`
`filter.collection.white`	表過濾白名單，多個表之間用`;`相隔，與黑名單不可同時使用。	`c1;c2`
`filter.collection.black`	表過濾黑名單，多個表之間用`;`相隔，與白名單不可同時使用。	`c1;c2`

若您希望使用基於 S3 快照的校正模式（推薦，詳見步驟二），還需配置以下PostgreSQL原生串連參數：

參數	說明	樣本值
`s3.export_state_file`	匯出狀態檔案路徑，由`nimo-shake`寫入，`nimo-full-check`讀取。	`../nimo-shake-s3-exports.json`
`s3.export_bucket`	DynamoDB資料匯出的目標S3儲存桶名稱。	`my-export-bucket`
`s3.export_prefix`	S3路徑首碼，用於組織匯出檔案。	`exports/my-project/`
`target.pg.endpoint_url`	PolarDB PostgreSQL版原生連接埠地址（`host:port`格式）。	`pc-xxx.pg.polardb.rds...:5432`
`target.pg.user`	PolarDB PostgreSQL版使用者名稱（同時作為schema名）。說明 DynamoDB 帳號本質上是普通資料庫帳號，其存取金鑰（SK）由您設定的密碼按相容演算法加密產生，因此 `target.pg.user` 與`target.access_key_id` 實際對應同一帳號，而 `target.pg.password` 是您的自訂密碼，SK 則是基於該密碼派生的密鑰。	`your-username`
`target.pg.password`	PolarDB PostgreSQL版密碼。	`your-password`
`target.pg.check_database`	存放增量校正表的專屬資料庫名。增量同步處理階段，`nimo-shake` 將變更記錄的主鍵寫入該庫的增量校正表，供 `s3` / `incr` 校正讀取。需在啟動前手動建立，`nimo-shake` 與 `nimo-full-check` 必須配置相同值。說明 `target.pg.check_database` 指向一個專門存放增量校正表的獨立資料庫，需在啟動同步前用高权限账号手動建立，並將其Owner指定為`target.pg.user`（即上面的 DynamoDB 帳號）。這樣該帳號才有許可權在庫內建立 schema 和增量校正表。	`nimoshake_check`

（可選）全量同步PostgreSQL直寫入模式

預設情況下，全量同步使用原生 PostgreSQL 直寫入模式，通過標準 PostgreSQL 協議直接將資料寫入PolarDB，繞過 DynamoDB 相容層 API，效能更高。如需回退到 DynamoDB 層 API（BatchWriteItem，每次最多寫入 25 條），可手動將該參數改為 dynamodb。可在 conf/nimo.conf 的 [SHAKE] 區塊中配置以下參數：

參數	說明	預設值
`full.write_protocol`	全量寫入協議。 `postgresql`（預設）：為PostgreSQL直寫高效能模式。 `dynamodb`：為DynamoDB層API（BatchWriteItem，單次最大 25 條）。	`postgresql`
`full.document.write.batch`	每次批量寫入的文檔數。PostgreSQL直寫時不受 25 條限制，可適當調大以進一步提升寫入吞吐。如切回 `dynamodb` 協議，請將此值改回 25。	`1000`
`full.document.concurrency`	每張表的並發寫入線程數。PostgreSQL直寫入模式下，工具會自動按此值設定串連池大小，確保線程與串連一一對應。	`4`

說明

PostgreSQL直寫入模式同樣需要配置 target.pg.* 四項參數（與s3 / incr 校正模式共用同一組配置，無需重複填寫）。
該模式僅影響全量同步階段的寫入路徑。增量同步處理始終通過 DynamoDB 相容層 API 寫入。

（可選）長時間全量同步的 Stream Buffer 保護

AWS DynamoDB Streams 中的變更記錄僅保留 24 小時。當源表資料量較大、預估全量同步耗時可能超過 24 小時時，全量階段產生的增量變更若在 Streams 中滯留超過 24 小時，將被 AWS 自動丟棄，導致增量同步處理丟資料。為此可開啟 Stream Buffer：增量監聽自全量同步開始即即時將 Streams 變更記錄落盤到本地緩衝，繞開 24 小時保留限制；待全量完成後再從本地緩衝消費回放，確保變更不丟失。該功能預設關閉，僅在預估全量耗時接近或超過 24 小時時按需開啟。可在 conf/nimo.conf 中配置以下參數：

參數	說明	預設值
`incr.stream_buffer.enable`	Stream Buffer 總開關。	`false`
`incr.stream_buffer.dir`	緩衝檔案目錄。留空則預設與 checkpoint 同級（`<workdir>/<id>-buffer`）。	空
`incr.stream_buffer.max_seg_size_mb`	單個緩衝段檔案大小上限（MB），超過即輪轉。	`64`
`incr.stream_buffer.max_seg_count`	單個緩衝段最大記錄數，超過即輪轉。	`10000`
`incr.stream_buffer.max_seg_duration_sec`	緩衝段強制輪轉的時間間隔（秒）。	`300`
`incr.stream_buffer.max_size_mb`	全域磁碟配額（MB），所有分區/表共用（非單分區）。需結合機器可用的磁碟規劃。	`51200`（50 GB）
`incr.stream_buffer.quota_backoff_sec`	緩衝超過全域配額後，寫入端每 30 秒記一條 ERROR 以提示營運擴容磁碟或加速全量同步。超過`quota_backoff_sec`秒後，同步程式自動終止。	`1800`

磁碟容量可按經驗公式粗估：所需磁碟 MB ≈ 全量耗時(秒) × 峰值寫入(條/秒) × (單條平均位元組 / 3) / 1MB × 1.5。磁碟務必從寬預留，寧可超配、不可不足。
全量同步期間讀取端不消費緩衝，因此一旦緩衝超過 max_size_mb 便無法自行排空—此時進程會在退避後崩潰，需擴大磁碟或提升全量同步速度後重啟。

根據您的作業系統架構，選擇對應的二進位檔案啟動同步任務。
說明
建議您在後台nohup運行，避免因終端服務斷開而導致同步任務中斷。
```
# 在NimoShake目錄下執行
nohup ./bin/nimo-shake.linux.amd64 -conf=./conf/nimo.conf > /dev/null 2>&1 &
```
程式將首先進行全量同步，完成後自動轉入增量同步處理階段，並持續運行。

步驟二：執行資料一致性校正

nimo-full-check與nimo-shake共用同一份conf/nimo.conf，無需切換目錄或編輯額外的設定檔。通過mode參數選擇校正模式：

scan：即時掃描源端與目標端資料進行對比。配置最簡單，無需額外基礎設施，但校正期間會持續消耗 DynamoDB 讀取配額。
s3（推薦）：全量同步完成時，nimo-shake 會自動將源端資料匯出至 S3，並產生一份時間點快照。nimo-full-check 以該快照為基準進行校正，無需再次掃描源端 DynamoDB。需在步驟一中配置 S3 相關參數。
incr（配合s3模式使用）：增量同步處理階段，nimo-shake 會將每條變更記錄的主鍵寫入 PolarDB 中的增量校正表（ct_{user}_{table}）。nimo-full-check 讀取增量校正表，僅對這些發生過變更的記錄進行定向校正。

兩種校正方式對比

	scan 模式	s3 / incr 模式
所需停機時間	較長。校正期間須停止業務寫入，避免源端資料在掃描過程中發生變化。	更短。基於快照校正，停機後僅需確認增量追平即可割接。
對源端的影響	持續消耗 DynamoDB 讀取配額。	全量校正基於 S3 快照，不消耗 DynamoDB 配額；增量校正（incr 模式）僅對有變更的記錄發起查詢，配額消耗較少。
配置複雜度	低。	較高，需額外配置S3和PolarDB PostgreSQL版串連。

（可選）增量持續校正（drain 模式，預設關閉）：在預設情況下，incr 模式是在割接停機視窗內一次性校正增量校正表中累積的全部變更記錄。當增量同步處理期間較長（數小時至數天）時，增量校正表會積累大量待校正主鍵，導致割接時的最終校正耗時偏長。開啟增量持續校正（drain）後，nimo-full-check mode=incr 將轉為常駐進程：在增量同步處理期間持續地“邊校正邊消費”增量校正表中已確認一致的記錄，使增量校正表在割接前始終維持在很小的規模；割接時僅需對極少量剩餘尾巴做一次最終校正即可，從而將停機視窗內的校正耗時大幅壓縮。該功能預設關閉，不影響現有任何行為；僅當增量周期長、對割接停機時間敏感時按需開啟。

重要

該開關 check_drain.enable 同時作用於 nimo-shake 與 nimo-full-check，二者共用同一份 conf/nimo.conf，配置一次即對兩個工具生效。
生產環境開啟前，建議先在測試執行個體上完成一次完整的流量驗證。

可在 conf/nimo.conf 中配置以下參數（預設情況下無需修改）：

參數	說明	預設值
`check_drain.enable`	增量持續校正（drain）總開關，同時作用於 `nimo-shake` 與 `nimo-full-check`。`false`（預設）時行為與現狀完全相同（`incr` 模式一次性跑完即退出）。	`false`
`check_drain.finalize_flag_file`	finalize 哨兵檔案路徑。割接時營運 `touch` 該檔案，常駐校正進程便在下一輪進入最終校正收尾、產出權威差異報告並退出。`enable=true` 時必填。	空
`check_drain.check_qps`	drain 校正每秒 `BatchGetItem` 上限（源、目標各自獨立），用於限制 drain 對源端/目標端的壓力，避免搶佔增量同步處理資源。若校正偏慢可適當調大。	`20`

說明

除上表外，conf/nimo.conf 中還有 check_drain.cycle_interval_sec、check_drain.settle_sleep_ms、check_drain.consistent_read 等進階調優參數，一般保持預設即可，含義見設定檔內注釋。
開啟 drain 後，割接階段不再是“再跑一次 nimo-full-check”，而是改為 touch 上述 finalize 哨兵檔案，觸發常駐進程收尾併產出最終差異報告。詳見執行業務割接-最後一次校正。
incr 模式的常駐進程應在 s3 基準校正跑完之後啟動，其餘校正流程與預設情況一致。

啟動校正任務。

nohup ./bin/nimo-full-check.linux.amd64 -conf=./conf/nimo.conf > /dev/null 2>&1 &

校正工具會將詳細日誌和資料差異報告分別存放在預設目錄logs/和nimo-full-check-diff/中。

[INFO] documentChecker[0] with table[{nimo-shake src2}] [success: 91, fail: 0] item matches!
[INFO] documentChecker[2] with table[{nimo-shake src1}] [success: 96, fail: 0] item matches!
[INFO] documentChecker[0] with table[{nimo-shake src2}] [success: 92, fail: 0] item matches!
[INFO] documentChecker[2] with table[{nimo-shake src1}] [success: 97, fail: 0] item matches!
[INFO] documentChecker[0] with table[{nimo-shake src2}] [success: 93, fail: 0] item matches!
[INFO] documentChecker[0] with table[{nimo-shake src2}] [success: 94, fail: 0] item matches!
[INFO] documentChecker[2] with table[{nimo-shake src1}] [success: 98, fail: 0] item matches!
[INFO] documentChecker[0] with table[{nimo-shake src2}] [success: 95, fail: 0] item matches!
[INFO] documentChecker[2] with table[{nimo-shake src1}] [success: 99, fail: 0] item matches!
[INFO] documentChecker[0] with table[{nimo-shake src2}] [success: 96, fail: 0] item matches!
[INFO] documentChecker[2] with table[{nimo-shake src1}] [success: 100, fail: 0] item matches!
[INFO] documentChecker[2] with table[{nimo-shake src1}] check completed. real total count: 100
[INFO] documentChecker[2] with table[{nimo-shake src1}] close executor
[INFO] documentChecker[2] finishes checking table[src1]
[INFO] documentChecker[0] with table[{nimo-shake src2}] [success: 97, fail: 0] item matches!
[INFO] documentChecker[0] with table[{nimo-shake src2}] [success: 98, fail: 0] item matches!
[INFO] documentChecker[0] with table[{nimo-shake src2}] [success: 99, fail: 0] item matches!
[INFO] documentChecker[0] with table[{nimo-shake src2}] [success: 100, fail: 0] item matches!
[INFO] documentChecker[0] with table[{nimo-shake src2}] check completed. real total count: 100
[INFO] documentChecker[0] with table[{nimo-shake src2}] close executor
[INFO] documentChecker[0] finishes checking table[src2]
[INFO] all documentCheckers finish
[INFO] checker finishes!
[INFO] full-check completes!

（可選）修複不一致資料：
1. 當源端DynamoDB與目前PolarDB PostgreSQL版叢集中的資料不一致時，將在指定的diff目錄中展示不一致的表，例如：
  說明
  diff目錄預設為nimo-full-check-diff，您可以在conf/nimo.conf中進行配置，參數為diff_output_file。
```
nimo-full-check-diff/
  └── testtable-0
  └── testtable-1
```
2. 執行修複任務：
```
nohup ./bin/nimo-repair.linux.amd64 -conf=./conf/nimo-repair.conf > /dev/null 2>&1 &
```

步驟三：（可選）配置反向同步

在準備進行業務割接前，您可以預先配置從PolarDB PostgreSQL版到源端DynamoDB的反向同步鏈路。該鏈路在業務正式切換至PolarDB期間啟動，用於實現資料的反向迴流，為業務復原提供保障。配置流程如下：

環境準備

確保PolarDB參數wal_level的值為logical。
建立高許可權資料庫帳號並授權：
1. （可選）如果您尚未建立高許可權帳號，請前往PolarDB控制台，在叢集的配置与管理 > 账号管理中建立高許可權帳號。

建立邏輯複製槽和發布：使用高許可權帳號串連至polardb_internal_dynamodb資料庫，執行以下SQL命令以建立一個邏輯複製槽，並向DynamoDB账号授予複製許可權，最終發布包含該資料庫下所有表的訂閱。您可以查看所建立邏輯複製槽的活躍狀態。

-- 建立邏輯複製槽，'flink_slot' 名稱需與後續 Flink 配置保持一致
SELECT * FROM pg_create_logical_replication_slot('flink_slot', 'pgoutput');

-- 為之前建立的 DynamoDB 專用帳號授予 REPLICATION 許可權
-- 將 <your_dynamodb_user> 替換為您的 DynamoDB 專用帳號名
ALTER ROLE <your_dynamodb_user> REPLICATION;

-- 檢查複製槽狀態，此時 flink_slot 的 active 應為 f (false)
SELECT * FROM pg_replication_slots;

開通並配置Flink：
1. 開通Realtime ComputeFlink版，並建立一個Flink工作空間。
  重要
  Flink工作空間需與PolarDB叢集位於同一個VPC下。
2. 為Flink工作空間配置公網訪問，使其能夠串連AWS DynamoDB。
配置PolarDB叢集的IP白名單：
1. 在Flink控制台，單擊工作空間的详情按鈕，在工作空間詳情頁面擷取其網段資訊。
2. 前往PolarDB控制台，在叢集的配置与管理 > 集群白名单中新增IP白名单分组，將Flink的網段資訊添加進去。
驗證PolarDB叢集與Flink工作空間連通性：
1. 在Flink控制台，進入工作空間，單擊右上方的網路探測表徵圖。
2. 填寫PolarDB叢集主節點的私人地址與連接埠，單擊探測。
3. 彈窗提示網路探測成功連通，即叢集白名單配置正確。

部署Flink作業

下載反向同步工具：PolarDBBackSync.jar。

準備設定檔：建立名為application.yaml的設定檔，內容如下：

snapshot:
  mode: never

source:
    # PolarDB主節點的私網地址
    hostname: pc-xxx.pg.polardb.rds.aliyuncs.com
    # PolarDB主節點的私網連接埠
    port: 5432
    # 之前建立的邏輯複製槽的名稱
    slotName: flink_slot

target:
  # 目標 AWS DynamoDB 的 region
  region: cn-north-1

# (可選) 表過濾配置，whiteTableSet 和 blackTableSet 只能聲明一個
filter:
  # whiteTableSet: 需要反向迴流的表
  # blackTableSet: 不需要反向迴流的表
  whiteTableSet:
  blackTableSet:

# Flink 作業的檢查點（checkpoint）間隔，單位毫秒
checkpoint:
  interval: 3000

上傳檔案：進入Flink控制台，找到並進入目標工作空間，在檔案管理頁面上傳PolarDBBackSync.jar和application.yaml。

安全儲存憑證：為避免明文暴露密鑰，建議使用Flink的變數管理功能儲存敏感資訊。在變數管理頁面新增以下四個變數：

變數名稱	變數值
`polardbusername`	PolarDB的DynamoDB帳號。
`polardbpassword`	PolarDB的DynamoDB帳號的原始密碼，並非DynamoDB帳號密鑰。
`dynamodbak`	AWS DynamoDB的AccessKey。
`dynamodbsk`	AWS DynamoDB的SecretKey。
`configfilename`	（可選）附加依賴檔案名稱，預設為`application.yaml`。

部署並啟動作業：

進入作業營運頁面，選擇部署作業 > JAR作業。

填寫以下主要參數，其他參數可根據業務環境進行配置。然後單擊部署。

參數名稱	填寫參考
部署模式	固定為流模式。
部署名稱	填寫作業部署名稱，此處以PolarDBBackSync為例。
引擎版本	固定為vvr-11.3-jdk11-flink-1.20。
JAR URI	選擇已上傳的`PolarDBBackSync.jar`。
Entry Point Class	固定為`org.example.PolarDBCdcJob`。
Entry Point Main Arguments	固定為： `--polardbusername ${secret_values.polardbusername}` `--polardbpassword ${secret_values.polardbpassword}` `--dynamodbak ${secret_values.dynamodbak}` `--dynamodbsk ${secret_values.dynamodbsk}` （可選）`--configfilename ${secret_values.configfilename}`
附加依賴檔案	選擇已上傳的`application.yaml`。

重要

如果您的密碼或其他參數值中包含特殊字元，可能會導致 Flink 作業解析參數失敗。為防止此問題，請在作業建立完成後，在部署詳情 > 運行參數配置 中點擊編輯按鈕，在其他配置中，添加以下配置來防止此類問題：env.java.opts: -Dconfig.disable-inline-comment=true。

部署成功後，單擊啟動 > 無狀態。

驗證與清理

驗證：在作業啟動後，使用高許可權帳號串連至PolarDB叢集的polardb_internal_dynamodb資料庫，並執行SELECT * FROM pg_replication_slots;。若複製槽flink_slot的active欄位變為t（true），則表示Flink作業已成功串連。此時即可在PolarDB叢集內開始匯入業務流量。
清理：當不再需要進行反向同步時，您可執行以下步驟以釋放相關資源節省費用。
- Realtime ComputeFlink版：
  - 停止作業：前往Flink控制台，在目標工作空間中，進入作業營運頁面，找到目標作業並單擊停止
  - 釋放執行個體：返回Flink控制台，找到目標工作空間，單擊釋放資源。
- PolarDB叢集：使用高許可權帳號串連至polardb_internal_dynamodb資料庫，執行以下命令刪除邏輯複製槽。
```
SELECT pg_drop_replication_slot('flink_slot');
```

步驟四：執行業務割接

增量同步處理延遲較低且資料一致性校正無差異後，可計劃業務割接。當您準備好進行最終的業務切換時，請遵循以下嚴謹的步驟：

最終校正：在計劃的停機視窗前，反覆運行一致性校正工具，直至確認增量同步處理延遲極低，且資料差異數量降至0或可接受的範圍內。
停止源端寫入：在停機視窗開始時，暫停所有向源端AWS DynamoDB寫入資料的業務應用。
等待同步完成：觀察nimo-shake的日誌，確認已無新的增量資料需要同步。
最後一次校正：再次運行nimo-full-check工具，確保源端和目標端的資料完全一致。
說明
若已開啟增量持續校正（drain 模式），此步驟無需再次啟動 nimo-full-check：常駐校正進程已在增量期間持續校正並消費增量校正表。待確認 nimo-shake 增量寫入速率歸零（隊列排空）後，建立 finalize 哨兵檔案（即 check_drain.finalize_flag_file 指向的檔案，如 touch /tmp/nimo-finalize），常駐進程會自動進入最終校正收尾、產出權威差異報告並退出。
停止同步工具：在確認資料完全同步後，停止nimo-shake進程。
切換應用串連：停止業務應用，將業務應用的資料庫連接配置，從AWS DynamoDB的地址切換為PolarDB的DynamoDB訪問地址。
（可選）開啟反向同步任務：在業務正式切換至PolarDB期間開啟反向同步任務，基於阿里雲Realtime ComputeFlink版實現資料的反向迴流，為業務復原提供保障。
啟動業務：重啟業務應用。至此，割接完成。
（可選）停止反向同步任務：在割接完成且業務穩定運行一段時間後，確認資料一致性滿足業務需求後，即可安全停止反向同步任務（停止Flink作業與釋放相關資源）。

附錄：為測試環境類比即時業務流量

如果您希望在測試環境中類比一個真實的、持續有資料寫入的遷移情境，可以使用以下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)
}