MaxCompute:Migrate data from RDS to MaxCompute based on dynamic partitioning

Step 1: Create the destination table

1. Log in to the DataWorks console.

2. In the left navigation pane, click Workspace.

3. In the Actions column of your workspace, click Shortcuts > Data Development.

4. Right-click the workflow and choose Create Table > MaxCompute > Table.

5. On the Create Table page, select the engine instance, schema, and path. Enter a Table Name and click Create.

6. On the table editing page, click the icon to switch to DDL mode.

7. In the DDL dialog box, paste the following statement and click Generate Table Schema. In the Confirm dialog box, click OK.

   CREATE TABLE IF NOT EXISTS ods_user_info_d (
   uid STRING COMMENT 'User ID',
   gender STRING COMMENT 'Gender',
   age_range STRING COMMENT 'Age range',
   zodiac STRING COMMENT 'Zodiac sign'
   )
   PARTITIONED BY (
   dt STRING
   );

8. Click Submit to Production Environment.

Step 2: Create an offline sync node

1. On the data analytics page, right-click the workflow and choose Create Node > Data Integration > Offline synchronization.

2. In the Create Node dialog box, enter a Name and click Confirm.

3. Configure the data source and destination:

   Field	Value
   Source	Select the MySQL data source you created
   Resource Group	Select an exclusive resource group for Data Integration
   Destination	Select the MaxCompute data source you created

4. Test connectivity, then click Next to configure the task.

   

Step 3: Configure partition parameters

1. In the right-side navigation pane, click Properties.

2. In the Scheduling Parameters section, review the default parameter. The default value is ${bizdate} in yyyymmdd format. This maps to the partition field in the Destination section and resolves to the previous day's date at runtime. To use a different date offset, replace ${bizdate} with a custom parameter. Wrap the date formula in square brackets []:

   The default calculation unit is days. For add_months, the unit is months: $[add_months(yyyymmdd,12*N)-M/24/60] computes (current timestamp − 12×N months − M/24/60 days) and returns the result in yyyymmdd format.

   Offset	Formula	Notes
   N days later	$[yyyymmdd+N]	Default unit is days
   N days earlier	$[yyyymmdd-N]	Default unit is days
   N weeks later	$[yyyymmdd+7*N]	Multiply by 7 for weeks
   N weeks earlier	$[yyyymmdd-7*N]	Multiply by 7 for weeks
   N months later	$[add_months(yyyymmdd,N)]	add_months unit is months
   N months earlier	$[add_months(yyyymmdd,-N)]	add_months unit is months
   N years later	$[add_months(yyyymmdd,12*N)]	12 months per year
   N years earlier	$[add_months(yyyymmdd,-12*N)]	12 months per year
   N hours later	$[hh24miss+N/24]	Result in hh24miss format
   N hours earlier	$[hh24miss-N/24]	Result in hh24miss format
   N minutes later	$[hh24miss+N/24/60]	Result in hh24miss format
   N minutes earlier	$[hh24miss-N/24/60]	Result in hh24miss format

Step 4: Run and verify

1. Click the icon to run the node.

2. After the run completes, open the Operation Log to confirm the data was written successfully.

3. On the MaxCompute client, run the following query to verify the partition was created and data was loaded:

   SELECT count(*) from ods_user_info_d where dt = 20180913;

Step 1: Set up the three-node workflow

Node 1 — Create and populate a temporary table

Create an SQL script node and run the following statements. This creates a non-partitioned temporary table and loads all data from the source MaxCompute table into it.

drop table if exists ods_user_t;
CREATE TABLE ods_user_t (
        dt STRING,
        uid STRING,
        gender STRING,
        age_range STRING,
        zodiac STRING);
-- Copy data from the source table into the temporary table.
insert overwrite table ods_user_t select dt,uid,gender,age_range,zodiac from ods_user_info_d;

Node 2 — Sync all data from ApsaraDB RDS to MaxCompute

Create an offline sync node named mysql_to_odps. Configure it to sync all data from ApsaraDB RDS to MaxCompute without setting any partition.

Node 3 — Dynamically partition the data into the destination table

Create an SQL script node and run the following statements. The INSERT OVERWRITE ... partition(dt) syntax tells MaxCompute to read the dt field from each row and write it to the matching partition, creating the partition if it does not exist.

drop table if exists ods_user_d;
-- Create the partitioned destination table.
CREATE TABLE ods_user_d (
    uid STRING,
    gender STRING,
    age_range STRING,
    zodiac STRING
)
PARTITIONED BY (
    dt STRING
);
-- Dynamic partitioning: MaxCompute reads the dt field from each row and
-- writes it to the matching partition. For example, a row where dt=20181025
-- is written to the partition dt=20181025, which is created automatically.
-- Include dt in the SELECT list so MaxCompute can use it as the partition key.
insert overwrite table ods_user_d partition(dt)
select dt,uid,gender,age_range,zodiac from ods_user_t;
-- Drop the temporary table to free up storage.
drop table if exists ods_user_t;

In MaxCompute, you can use SQL statements to complete data synchronization.For more information about SQL statements, see Learning about MaxCompute, a big data tool of Alibaba Cloud--Use of partitioned tables.

Step 2: Configure the workflow dependencies

Set the three nodes to run sequentially: Node 1 → Node 2 → Node 3.

Step 3: Run and verify

1. Trigger the workflow. Monitor the execution progress, paying attention to the dynamic partitioning step in Node 3.

   

2. After the workflow completes, run the following query on the MaxCompute client to verify data in a specific partition:

   SELECT count(*) from ods_user_d where dt = 20180913;