MaxCompute:Migrate RDS to MaxCompute with dynamic partitioning

Login DataWorks console.

Create a destination table in MaxCompute.

1. In the left-side navigation pane, click Workspace.

2. In the Actions column of your workspace, click Quick Access > Data Development.

3. Right-click the workflow you created and choose New table > MaxCompute > Table.

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

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

6. In the DDL dialog box, enter the following statement and click Generate Table Schema. In the Confirm operation dialog box that appears, 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
   );                           

7. Click Submit to Production Environment.

Create an offline sync node.

1. Go to the data analytics page. Right-click the specified workflow and choose Create Node > Data Integration > Offline synchronization.

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

3. Select the data source, resource group, and destination, and then test the connection.

   • Source: The MySQL data source that you created.

   • Resource Group: Select an exclusive resource group for Data Integration.

   • Destination: The MaxCompute data source that you created.

4. Click Next to configure the task. For the source, select the ods_user_info_d table and set the split key to uid. For the destination, select the ods_user_info_d table and enter dt=${bizdate} for the partition information. Set the Tunnel resource group to public transport resource and the schema to default. For the write mode, select Insert Overwrite (Clear existing data before writing), and for writing empty strings as null, select No.

Configure scheduling parameters.

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

2. In the Scheduling Parameter section, the default parameter is ${bizdate}, which uses the yyyymmdd format.

   Note

   The default parameter value corresponds to the Destination configured for the Partition Information. When the task runs, the system automatically replaces the partition value with the business date (the day before the task's execution), as ETL jobs typically process the previous day's data. If you need to use the current run date, you must customize the parameter.

   Custom parameter settings: You can choose a specific date and format, as shown in the following examples:

   • N years later: $[add_months(yyyymmdd,12*N)]

   • N years earlier: $[add_months(yyyymmdd,-12*N)]

   • N months earlier: $[add_months(yyyymmdd,-N)]

   • N weeks later: $[yyyymmdd+7*N]

   • N months later: $[add_months(yyyymmdd,N)]

   • N weeks earlier: $[yyyymmdd-7*N]

   • N days later: $[yyyymmdd+N]

   • N days earlier: $[yyyymmdd-N]

   • N hours later: $[hh24miss+N/24]

   • N hours earlier: $[hh24miss-N/24]

   • N minutes later: $[hh24miss+N/24/60]

   • N minutes earlier: $[hh24miss-N/24/60]

   Note

   • Use square brackets ([]) to define the calculation formula for a custom variable. For example: key1=$[yyyy-mm-dd].

   • By default, the calculation unit for custom variables is days. For example, $[hh24miss-N/24/60] represents the result of (yyyymmddhh24miss - (N/24/60 * 1 day)), formatted as hh24miss.

   • The calculation unit for the add_months function is months. For example, $[add_months(yyyymmdd,12*N)-M/24/60] represents the result of (yyyymmddhh24miss - (12 * N * 1 month) - (M/24/60 * 1 day)), formatted as yyyymmdd.

Click icon to run the code.

You can Runtime Log view the results.

Filter historical data from the ApsaraDB RDS source by date.

Perform a data backfill. For more information, see Manage data backfill instances.

In the run log, check the extraction results for the ApsaraDB RDS data.

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%

Verify the result. Run the following command in the MaxCompute client to check if the data was written.

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

Synchronize all data to a temporary table in MaxCompute and create an SQL script node. Run the following command.

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

Create a synchronization task node named mysql_to_odps. This is a simple task to synchronize all data from RDS to MaxCompute. No partitions are required.

Use an SQL statement to perform dynamic partitioning into the target table. The command is as follows.

drop table if exists ods_user_d;
//Create a partitioned ODPS table. This is the final destination table.
    CREATE TABLE ods_user_d (
    uid STRING,
        gender STRING,
        age_range STRING,
        zodiac STRING
)
PARTITIONED BY (
    dt STRING
);
//Run the dynamic partitioning SQL statement. It automatically partitions data based on the dt field of the temporary table. Records with the same value in the dt field are placed in a partition created for that value.
//For example, if some records have the value 20181025 in the dt field, a partition dt=20181025 is automatically created in the ODPS partitioned table.
//The dynamic partitioning SQL is as follows.
//Note that the dt field is included in the SELECT statement. This specifies that partitions are automatically created based on this field.
insert overwrite table ods_user_d partition(dt)select dt,uid,gender,age_range,zodiac from ods_user_t;
//After the import is complete, delete the temporary table to save storage costs.
drop table if exists ods_user_t;

Configure the three nodes into a workflow to run in sequence.

View the execution process. Focus on the dynamic partitioning process of the last node.

=20181203065434115g3a2eqsa
Log view:
http://logview.odps.aliyun.com/logview/?h=http://service.odps.aliyun.com/api&p=DataWorks_DOC&i=20181203065434115g3a2eqsa&token=V0NaNDFxUmpn...
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:
    ...

Verify the result. Run the following command in the MaxCompute client to check the data that was written.

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