使用DataWorks將RDS資料移轉至MaxCompute實現動態分區-雲原生MaxCompute MaxCompute-阿里雲 - MaxCompute

本文為您介紹如何使用DataWorksData Integration同步功能自動建立分區，動態地將RDS中的資料移轉至MaxComputeMaxCompute。

前提條件

準備DataWorks環境
1. 開通MaxCompute。
2. 在DataWorks上完成建立商務程序，本例使用DataWorks簡單模式。詳情請參見建立商務程序。
新增資料來源
- 新增MySQL資料來源作為資料來源，詳情請參見配置MySQL資料來源。
- 新增MaxCompute資料來源作為目標資料來源接收RDS資料，詳情請參見配置MaxCompute資料來源。

自動建立分區

準備工作完成後，需要將RDS中的資料定時每天同步到MaxCompute中，自動建立按天日期的分區。詳細的資料同步任務的操作和配置請參見DataWorks資料開發和營運。

說明

本例使用DataWorks簡單模式，建立工作空間時，預設保持參加資料開發（Data Studio）公測不開啟，公測工作空間不適用本例。

登入DataWorks控制台。
在MaxCompute上建立目標表。
1. 在左側導覽列，單擊工作空間。
2. 單擊相應工作空間操作列的快速進入 > 数据开发。
3. 按右鍵已建立的商務程序，選擇新建表 > MaxCompute > 表。
4. 在新建表頁面，選擇引擎執行個體、Schema和路徑後，輸入表名，並單擊新建。
5. 在表的編輯頁面，單擊表徵圖，切換為DDL模式。
6. 在DDL對話方塊，輸入如下建表語句，單擊生成表結構。然後在確認操作對話方塊裡單擊确认。
```
CREATE TABLE IF NOT EXISTS ods_user_info_d (
uid STRING COMMENT '使用者ID',
gender STRING COMMENT '性別',
age_range STRING COMMENT '年齡段',
zodiac STRING COMMENT '星座'
)
PARTITIONED BY (
dt STRING
);                           
```
7. 單擊提交到生產環境。
建立離線同步節點。
1. 進入資料開發頁面，按右鍵指定商務程序，選擇建立節點 > Data Integration > 離線同步。
2. 在建立節點對話方塊中，輸入節點名稱，並單擊確認。
3. 選擇資料來源、資源群組和資料去向，並測試連通性。
  - 數據來源：已建立的MySQL資料來源。
  - 我的資源組：選擇獨享Data Integration資源群組。
  - 數據去向：已建立的MaxCompute資料來源。
4. 單擊下一步，配置任務。資料來源配置：表選擇ods_user_info_d，切分鍵填寫uid。資料去向配置：Tunnel資源群組選擇公用傳輸資源，schema為default，表選擇ods_user_info_d，分區資訊填寫dt=${bizdate}，寫入模式選擇寫入前清理已有資料(Insert Overwrite)，Null 字元串轉為Null寫入選擇否。
配置分區參數。
1. 在右側導覽列上，單擊調度配置。
2. 在調度參數地區，設定參數。參數值預設為系統內建的時間參數${bizdate}，格式為yyyymmdd。
  說明
  預設參數值與數據去向中的分区信息值對應。調度執行遷移任務時，目標表的分區值會被自動替換為任務執行日期的前一天，預設情況下，您會在當前執行前一天的業務資料，這個日期也叫做業務日期。如果您需要使用當天任務啟動並執行日期作為分區值，則需自訂參數值。
  自訂參數設定：使用者可以自主選擇某一天和格式配置，如下所示：
  - 後N年：$[add_months(yyyymmdd,12*N)]
  - 前N年：$[add_months(yyyymmdd,-12*N)]
  - 前N月：$[add_months(yyyymmdd,-N)]
  - 後N周：$[yyyymmdd+7*N]
  - 後N月：$[add_months(yyyymmdd,N)]
  - 前N周：$[yyyymmdd-7*N]
  - 後N天：$[yyyymmdd+N]
  - 前N天：$[yyyymmdd-N]
  - 後N小時：$[hh24miss+N/24]
  - 前N小時：$[hh24miss-N/24]
  - 後N分鐘：$[hh24miss+N/24/60]
  - 前N分鐘：$[hh24miss-N/24/60]
  說明
  
  使用中括弧（[]）編輯自訂變數參數的取值計算公式，例如 key1=$[yyyy-mm-dd]。
  
  預設情況下，自訂變數參數的計算單位為天。例如 $[hh24miss-N/24/60] 表示 (yyyymmddhh24miss-(N/24/60 * 1天)) 的計算結果，然後按 hh24miss 的格式取時分秒。
  
  使用add_months的計算單位為月。例如 $[add_months(yyyymmdd,12 N)-M/24/60] 表示 (yyyymmddhh24miss-(12 * N * 1月))-(M/24/60 * 1天) 的結果，然後按 yyyymmdd 的格式取年月日。
單擊表徵圖運行代碼。
您可以在作業記錄查看運行結果。

補資料實驗

如果您的資料中存在大量運行日期之前的歷史資料，需要實現自動同步和自動分區。您可以通過DataWorks的運維中心，選擇當前的同步資料節點，使用補數據功能實現。

在RDS端按照日期篩選出歷史資料。

您可以在同步節點數據來源地區設定數據過濾條件（例如設定為${bizdate}）。
執行補資料操作。詳情請參見補資料執行個體營運。

在啟動並執行日誌中查看對RDS資料的抽取結果。

從運行結果中可以看到MaxCompute已自動建立分區。作業記錄顯示 MySQL 到 ODPS 的 DataX 同步任務配置，其中 Reader 的 where=[20180913] 與 Writer 的 partition=[dt=20180913] 相對應，確保讀取條件與寫入分區日期一致。

Alibaba DI Console, Build 201805310000 .
Copyright 2018 Alibaba Group, All rights reserved .
Start Job[16961870], traceId [283789484710656#79023#None#None#228255635341196741#None#None#rds_sync], running in Pipeline[basecommon_
89484710656]
The Job[16961870] will run in PhysicsPipeline [basecommon_group_283789484710656_oxs] with requestId [4f44180d-300c-47c3-8ea3-805d2
2018-12-02 03:31:25 : ---
Reader: mysql
                column=["uid","gender","age_range","zodiac"]
                connection=[{"datasource":"xxx","table":["`ods_user_info_d`"]}]]
                where=[20180913                          ]
                splitPk=[uid                             ]
Writer: odps
                isCompress=[false                        ]
                partition=[dt=20180913                    ]
                truncate=[true                           ]
                datasource=[odps_first                   ]
                column=["uid","gender","age_range","zodiac"]
                emptyAsNull=[false                       ]
                table=[ods_user_info_d                    ]
Setting:
                errorLimit=[{"record":""}                 ]
                speed=[{"concurrent":1,"dmu":1,"mbps":"10","throttle":true}]
2018-12-02 03:31:26 : State: 1(SUBMIT) | Total: 0R 0B | Speed: 0R/s 0B/s | Error: 0R 0B | Stage: 0.0%
2018-12-02 03:31:36 : State: 3(RUN) | Total: 0R 0B | Speed: 0R/s 0B/s | Error: 0R 0B | Stage: 0.0%

運行結果驗證。在MaxCompute用戶端執行如下命令，查看資料寫入情況。
```
SELECT count(*) from ods_user_info_d where dt = 20180913;
```

Hash實現非日期欄位分區

如果您的資料量較大，或沒有按照日期欄位對第一次全量的資料進行分區，而是按照省份等非日期欄位分區，則此時Data Integration操作將不能實現自動分區。這種情況下，您可以按照RDS中某個欄位進行Hash，將相同的欄位值自動存放到這個欄位對應值的MaxCompute分區中。

將資料全量同步到MaxCompute的一個暫存資料表，建立一個SQL指令碼節點。執行如下命令。

drop table if exists ods_user_t;
CREATE TABLE ods_user_t ( 
        dt STRING,
        uid STRING,
        gender STRING,
        age_range STRING,
        zodiac STRING);
--將MaxCompute表中的資料存入暫存資料表。
insert overwrite table ods_user_t select dt,uid,gender,age_range,zodiac from ods_user_info_d;

建立同步任務的節點mysql_to_odps，即簡單的同步任務。將RDS資料全量同步到MaxCompute，無需設定分區。

使用SQL語句進行動態分區到目標表，命令如下。

drop table if exists ods_user_d;
//建立一個ODPS分區表（最終目的表）。
    CREATE TABLE ods_user_d (
    uid STRING,
        gender STRING,
        age_range STRING,
        zodiac STRING
)
PARTITIONED BY (
    dt STRING
);
//執行動態分區SQL，按照暫存資料表的欄位dt自動分區，dt欄位中相同的資料值，會按照這個資料值自動建立一個分區值。
//例如dt中有些資料是20181025，會自動在ODPS分區表中建立一個分區，dt=20181025。
//動態分區SQL如下。
//可以注意到SQL中select的欄位多寫了一個dt，就是指定按照這個欄位自動建立分區。
insert overwrite table ods_user_d partition(dt)select dt,uid,gender,age_range,zodiac from ods_user_t;
//匯入完成後，可以把暫存資料表刪除，節約儲存成本。
drop table if exists ods_user_t;

在MaxCompute中您可以通過SQL陳述式完成資料同步。

將三個節點配置成一個工作流程，按順序執行。

執行順序依次為暫存資料表、mysql_to_odps、目標表。

查看執行過程。您可以重點觀察最後一個節點的動態分區過程。

=20181203065434115g3a2eqsa
Log view:
http://logview.odps.aliyun.com/logview/?h=http://service.odps.aliyun.com/api&p=DataWorks_DOC&i=20181203065434115g3a2eqsa&token=V0NaNDFxUmpn...
Job Queueing...
Summary:
resource cost: cpu 0.00 Core * Min, memory 0.00 GB * Min
inputs:
  dataworks_doc.ods_user_t: 20028 (119496 bytes)
outputs:
  dataworks_doc.ods_use xxx 20028 (119176 bytes)
Job run time: 0.000
Job run mode: service job
Job run engine: execution engine
M1:
  instance count: 1
  run time: 0.000
  instance time:
    min: 0.000, max: 0.000, avg: 0.000
  input records:
    ...

運行結果驗證。在MaxCompute用戶端執行如下命令，查看資料寫入情況。
```
SELECT count(*) from ods_user_d where dt = 20180913;
```