如何將DynamoDB的數據增量遷移到表格存儲

Amazon DynamoDB是一個徹底託管的NoSQL數據庫服務，能夠提供快速的、可預期的性能，而且能夠實現無縫擴展。因爲DynamoDB並能夠根據實際需求對錶進行擴展和收縮，這個過程既不須要中止對外服務，也不會下降服務性能，一經推出就收到了廣大AWS用戶的歡迎。

一樣，表格存儲是構建在阿里雲飛天分佈式系統之上的分佈式NoSQL數據庫服務。做爲同 DynamoDB 很是類似的 __雲NoSQL數據庫服務__，表格存儲的自動負載均衡機制能夠自動對錶進行擴展，實現數據規模與訪問併發上的無縫擴展，提供海量結構化數據的存儲和實時訪問。

表格存儲可使用戶把操做和擴展分佈式數據庫的沉重負擔，交給阿里雲來處理，這樣，用戶就不須要擔憂硬件配置、磁盤故障、機器故障、軟件安裝和升級等工做，能夠更專一到業務邏輯中去。

今天，就給你們介紹如何將DynamoDB的數據增量遷移到表格存儲。

數據轉換規則

表格存儲支持的數據格式有：

• String - 可爲空，可爲主鍵，爲主鍵列時最大爲 1 KB，爲屬性列時爲2MB。
• Integer - 64 bit，整型，可爲主鍵，8 Bytes。
• Binary - 二進制數據，可爲空，可爲主鍵，爲主鍵列時最大爲 1 KB，爲屬性列時爲2MB。
• Double - 64 bit，Double 類型，8 Bytes。
• Boolean - True/False，布爾類型，1 Byte。

目前 DynamoDB 支持多種數據格式：

• 標量類型 - 標量類型可準確地表示一個值。標量類型包括數字、字符串、二進制、布爾值和 null。
• 文檔類型 - 文檔類型可表示具備嵌套屬性的複雜結構 - 例如您將在 JSON 文檔中找到的結構。文檔類型包括列表和映射。
• 集類型 - 集類型可表示多個標量值。集類型包括字符串集、數字集和二進制集。

因爲DynamoDB支持文檔型數據類型，咱們須要將文檔型轉換爲一個String類型或者Binary類型存儲到表格存儲中，在讀取時須要反序列化成Json。

故，從DynamoDB遷移到表格存儲時，咱們作以下的數據轉換：

DynamoDB類型	數據示例	TableStore對應類型
id (N)	'123'	Integer
level (N)	'2.3'	Double, 不能爲主鍵
afea (NULL)	TRUE	String,空字符串
binary (B)	0x12315	binary
binary_set (BS)	{ 0x123, 0x111 }	binary
bool (BOOL)	TRUE	boolean
list (L)	[ { "S" : "a" }, { "N" : "1" }]	string
map (M)	{ "key1" : { "S" : "value1" }}	string
str (S)	This is test！	string
num_set (NS)	{ 1, 2 }	string
str_set (SS)	{ "a", "b" }	string

增量實現機制

咱們使用DynamoDB的Stream數據流獲取DynamoDB表中的增刪改操做，將操做同步到表格存儲中，爲了不環境搭建，將同步程序運行在Lambda 中，流程以下圖：

使用Stream數據流中的'eventName'字段來判別數據的增刪改操做：

• "INSERT"： 插入數據，對應PutRow

• "MODIFY" ： 修改數據

  "REMOVE"：刪除數據，對應 DeleteRow
  • 若是OldImage 與 NewImage的key相同，則爲更新數據，對應Update
  • 若OldImage的Key 數量大於 NewImage的Key數量， 則爲刪除數據，將二者差集的keys作刪除，對應Delete

須要特別注意的是：

1. 上述 Stream 中增刪改操做轉換行爲符合業務的指望。
2. 表格存儲目前還不支持二級索引，故只能同步主表的數據。
3. DynamoDB 中表的主鍵同TableStore中的主鍵保持一致，且數字類型的主鍵只能爲整型。
4. DynamoDB 對單個項目的大小限制爲400KB，表格存儲中單行雖然沒有限制，但一次提交的數據量不能超過4MB。 DynamoDB限制項參考 及 TableStore 限制項參考
5. 若是先進行全量數據遷移，則須要在全量遷移以前開啓 Stream。因爲 DynamoDB Stream 只能保存最近24小時數據，故全量數據須要在24小時內遷移完成，在全量遷移完成後才能開啓 Lambda 的遷移任務。
6. 數據須要保證最終一致性。增量數據在同步時，可能會有對全量數據的重複寫入，好比 T0 時刻開啓 Stream 並進行全量遷移，T1 時刻完成，那麼 T0 到 T1 之間的時間段內的DynamoDB 數據操做會同步寫入到表格存儲中。

操做過程

1. 在DynamoDB中建立數據表

咱們以表Source爲例，主鍵爲user_id(字符串類型)，排序鍵爲action_time(數字)。因爲DynamoDB的預留設置會影響讀寫的併發，故須要注意預留的設置。

2. 開啓source 表的Stream

Stream模式須要爲： 新舊映像 - 新舊項目的映像

3. 轉到Lambda的控制檯，建立相關的數據同步函數

實例函數名稱爲：data-to-table， 運行語言選擇爲 Python 2.7，使用 lambda-dynamodb-execution-role的角色。

4.關聯Lambda的事件源

點擊事件源的DynamoDB圖標，進行事件源配置，選擇 source 數據表批處理大小先選擇爲10進行小批量驗證，在實際運行過程當中建議爲100，因爲表格存儲的Batch操做最大爲200條數據，故不能超過200。

5. 配置Lambda的函數。

點擊 Lambda的函數圖標，進行函數相關的配置。

因爲tablestore須要依賴SDK及 protocolbuf等依賴包，咱們按照建立部署程序包 (Python)的方式進行 SDK依賴安裝及打包。

使用的函數zip包爲：lambda_function.zip 點擊下載 能夠直接本地上傳，也能夠先上傳到S3。

處理程序入口爲默認的 lambda_function.lambda_handler

基本設置中須要將超時事件設置在1分鐘以上（考慮到批量提交的延時及網絡傳輸時間）。

6. 配置Lambda的運行變量

在數據導入時，須要 TableStore 實例名、AK等相關信息，咱們可使用一下兩種方式：

• 方案一（推薦）：直接在Lambda 中配置相關的環境變量，以下圖.
  使用 Lambda的環境變量將使得同一函數代碼zip包可以靈活的支持不一樣的數據表，而不須要爲每一個數據源修改代碼包中的配置文件。

參考：Lambda環境變量說明

• 方案二： 也能夠打開 lambda_function.zip 修改其中的example_config.py，再打包上傳，或者上傳以後在控制檯上進行修改。
  

配置說明：

環境變量	必選	意義
OTS_ID	是	訪問表格存儲的AccessKeyId信息
OTS_SECRET	是	訪問表格存儲的AccessKeySecret信息
OTS_INSTANCE	是	導入的表格存儲的實例名稱
OTS_ENDPOINT	否	導入的表格存儲的域名，若是不存在，則使用默認的實例公網域名
TABLE_NAME	是	導入的表格存儲的表名
PRIMARY_KEY	是	導入的表格存儲的表的主鍵信息，須要保證主鍵順序，主鍵名稱須要同源表保持一致

特別注意：

1. 相同的變量名稱，優先會從Lambda中變量配置中讀取，若是不存在，則會從 example_config.py中讀取。
2. 因爲AK信息表明這資源的訪問權限，強烈建議使用只具備表格存儲特定資源寫權限的子帳號的AK，避免AK泄露帶來的風險，使用參考

7. 在表格存儲中建立數據表。

在表格存儲控制檯上建立數據表：__target__，主鍵爲 user_id（字符串）和action_time(整型)。

8. 測試調試。

在lambda控制檯上編輯事件源進行調試。

點擊右上角的 配置測試事件，輸入示例事件的json內容。
咱們準備了兩個示例的 Stream示例事件：

1. test_data_put.json 模擬向DynamoDB中插入一條數據的事件，查看文件
2. test_data_update.json 模擬向DynamoDB中更新一條數據的事件，查看文件
3. test_data_update.json 模擬向DynamoDB中刪除一條數據的事件，查看文件

咱們將上述三個事件的內容依次保存爲putdata、updatedata、deletedata。

保存以後，選擇須要使用的事件，點擊測試：

執行結果提示成功的話，則在表格存儲的 target表中就能夠讀到以下的測試數據。

依次選擇putdata、updatedata和deletedata，會發現表格存儲中的數據也會隨之更新和刪除。

9.正式運行

測試經過以後，咱們在DynamoDB中新寫入一條數據，在表格存儲中立刻就能夠讀到這條數據，以下圖。

10. 問題調查

Lambda 運行的日誌都會寫入到 CloudWatch 中，在 CloudWatch 選擇對應的函數名，則能夠實時查詢到 Lambda 的運行狀態。

代碼解析

Lambda函數中，主要的代碼邏輯爲lambda_function.py 查看代碼，其餘則爲表格存儲SDK的依賴。lambda_function.py中主要包含了一下幾個function：

• def batch_write_row(client, put_row_items) - 將組合好的數據 Item （包括增刪改）批量寫到表格存儲中
• def get_primary_key(keys) – 根據變量PRIMARY_KEY 拿到源表和目的表的主鍵信息。
• def generate_update_attribute(new_image, old_image, key_list) – 解析Stream中的Modify操做，是對部分屬性列的更新仍是刪除了部分屬性列。
• def generate_attribute(new_image, key_list) – 獲取單個Record中插入的屬性列信息。
• def get_tablestore_client() – 根據變量中的實例名、AK信息等初始化表格存儲的客戶端。
• def lambda_handler(event, context) – Lambda的入口函數。

若是有更復雜的同步邏輯，也能夠基於 lambda_function.py 進行修改。

lambda_function.py 中打印的狀態日誌沒有區分 INFO 或者 ERROR，爲了保證數據同步的一致性，還須要對日誌進行處理，並監控運行狀態或者使用 lambda 的錯誤處理機制保證對異常狀況的容錯處理。