Incremental data synchronization from ApsaraDB RDS to MaxCompute

Usage notes

• Incremental synchronization is not available for all data sources, such as HBase and OTSStream. To check if a specific data source supports incremental synchronization, see the documentation for its corresponding Reader plugin.

• The parameters required for incremental synchronization may vary between Reader plugins. For detailed parameter configurations, see the documentation for the specific Reader plugin. For more information, see Supported data sources and read/write plugins.

• For more information about configuring incremental synchronization, see Scenario: Configure an offline task for incremental synchronization.

• Synchronizing source data to MaxCompute external tables is not supported.

Incremental synchronization of static data

Because this type of data does not change after it is generated, you can easily partition the table based on how the data is created. A common approach is to partition by date, creating one partition per day.

1. In your ApsaraDB RDS database, run the following SQL statements to prepare the data.

   drop table if exists oplog;
   create table if not exists oplog(
   optime DATETIME,
   uname varchar(50),
   action varchar(50),
   status varchar(10)
    );
   Insert into oplog values(str_to_date('2016-11-11','%Y-%m-%d'),'LiLei','SELECT','SUCCESS');
   Insert into oplog values(str_to_date('2016-11-12','%Y-%m-%d'),'HanMM','DESC','SUCCESS');

   The two records above are considered historical data. You must first perform a full data synchronization to load the historical data into the previous day's partition.

2. In Data Studio, right-click Table under your workflow and select Create Table.

3. In the Create Table dialog box, enter ods_oplog for Table Name and click Submit.

4. Double-click the ods_oplog table. On the editor page that appears on the right, click DDL Statement and enter the following DDL statement.

   -- Create a MaxCompute table partitioned by day.
   create table if not exists ods_oplog(
    optime datetime,
    uname string,
    action string,
    status string
   ) partitioned by (ds string);

5. Configure a synchronization task for the historical data. For more information, see codeless UI configuration.

   After the synchronization task runs successfully, go to the node editor. On the Scheduling tab on the right, select Suspend scheduling, and then commit or deploy the node again. This prevents the task from running automatically.

6. Run the following statements to insert new records into the source ApsaraDB RDS table. These records will serve as the incremental data.

   insert into oplog values(CURRENT_DATE,'Jim','Update','SUCCESS');
   insert into oplog values(CURRENT_DATE,'Kate','Delete','Failed'); 
   insert into oplog values(CURRENT_DATE,'Lily','Drop','Failed');

7. Configure a task to synchronize the incremental data.

   In the Data Source section, set the filter to date_format(optime,'%Y%m%d')=${bdp.system.bizdate}. In the Data going section, set the Partition Key Column to ${bdp.system.bizdate}.

   Note

   This filter ensures that when the synchronization runs on November 15, it queries all new data from November 14 in the source table and writes it to the destination table's incremental partition.

8. View the synchronization result.

   On the Scheduling tab of the node editor, set the scheduling cycle to daily. After you commit or deploy the task, it runs automatically the next day. Once it succeeds, you can view the data in the destination MaxCompute table.

Incremental synchronization of continuously updated data

To track historical changes in a data warehouse, a common practice is to run a daily full synchronization of frequently updated tables, such as user or order tables. This approach saves a complete snapshot of the data each day, making it easy to access both historical and current states.

However, some real-world scenarios may require you to perform only incremental synchronization each day. Because MaxCompute does not support the UPDATE statement, you must use other methods to handle updated records. The following sections describe two strategies: daily full synchronization and daily incremental synchronization.

1. Run the following SQL statements to prepare the data.

   drop table if exists user ;
   create table if not exists user(
       uid int,
       uname varchar(50),
       deptno int,
       gender VARCHAR(1),
       optime DATETIME
       );
   -- Historical data
   insert into user values (1,'LiLei',100,'M',str_to_date('2016-11-13','%Y-%m-%d'));
   insert into user values (2,'HanMM',null,'F',str_to_date('2016-11-13','%Y-%m-%d'));
   insert into user values (3,'Jim',102,'M',str_to_date('2016-11-12','%Y-%m-%d'));
   insert into user values (4,'Kate',103,'F',str_to_date('2016-11-12','%Y-%m-%d'));
   insert into user values (5,'Lily',104,'F',str_to_date('2016-11-11','%Y-%m-%d'));
   -- Incremental data
   update user set deptno=101,optime=CURRENT_TIME  where uid = 2; -- Change NULL to a non-NULL value.
   update user set deptno=104,optime=CURRENT_TIME  where uid = 3; -- Change a non-NULL value to another non-NULL value.
   update user set deptno=null,optime=CURRENT_TIME  where uid = 4; -- Change a non-NULL value to NULL.
   delete from user where uid = 5;
   insert into user(uid,uname,deptno,gender,optime) values (6,'Lucy',105,'F',CURRENT_TIME);

2. Synchronize the data.

   • Daily full synchronization

     1. Run the following statement to create the MaxCompute table. For more information, see Create Table.

        -- Full synchronization
        create table ods_user_full(
            uid bigint,
            uname string,
            deptno bigint,
            gender string,
            optime DATETIME 
        ) partitioned by (ds string);

     2. Configure a full synchronization task.

        Note

        Because a full synchronization is required every day, you must set the scheduling cycle for the task to daily.

     3. Run the task and view the result in the destination MaxCompute table.

        With daily full synchronization, there is no distinction between a full and an incremental load. After the task runs automatically on the following day, you can view the complete, updated data.

   • Daily incremental synchronization

     This method is not recommended unless your environment does not support the DELETE statement or you cannot view deleted records by using SQL. In most cases, deletions are handled as a logical delete by converting DELETE operations into UPDATE operations. However, this approach may not support certain business requirements and can lead to data inconsistencies. Additionally, it requires a separate step to merge the incremental data with the existing historical data.

     Prepare the data

     You need to create two tables: a result table to store the current data and an incremental data table to hold the daily changes.

     -- Result table
     create table dw_user_inc(
         uid bigint,
         uname string,
         deptno bigint,
         gender string,
         optime DATETIME 
     );

     -- Incremental data table
     create table ods_user_inc(
         uid bigint,
         uname string,
         deptno bigint,
         gender string,
         optime DATETIME 
     )

     1. Configure a synchronization task to perform an initial full load of the data directly into the result table.

        Note

        This task only needs to run once. After it succeeds, click Scheduling on the right side of the page and select Suspend scheduling.

     2. Configure a synchronization task to write the incremental data into the incremental data table. Set the data filter, which is the where parameter, to date_format(optime,'%Y%m%d')=${bdp.system.bizdate}.

     3. Run the following statement to merge the data.

        insert overwrite table dw_user_inc 
        select 
        -- If a record exists in the incremental table (ODS), it indicates a change, so its data takes precedence.
        case when b.uid is not null then b.uid else a.uid end as uid,
        case when b.uid is not null then b.uname else a.uname end as uname,
        case when b.uid is not null then b.deptno else a.deptno end as deptno,
        case when b.uid is not null then b.gender else a.gender end as gender,
        case when b.uid is not null then b.optime else a.optime end as optime
        from 
        dw_user_inc a 
        full outer join ods_user_inc b
        on a.uid  = b.uid ;

        After you view the result, you will notice that the record marked for deletion was not synchronized correctly.

   The advantage of daily incremental synchronization is the smaller volume of data transferred each day. However, it can lead to data inconsistencies and requires additional computation to merge the data.

   We recommend using daily full synchronization for continuously updated data unless your use case strictly requires incremental updates. If you want to retain historical data for only a specific period, you can configure the table's lifecycle to automatically delete data that exceeds the retention period.