Incrementally pull data with a resource function

Video tutorial

Resources and data samples

• Simple Log Service resources

  • Project: client-project

  • Source Logstore: client-log

    The sample data includes the following fields: __tag__:__client_ip__ (client IP address), c_id (customer ID), staff_id (staff ID), status (follow-up status, such as "Ongoing follow-up" and "Follow-up"), and tag (customer tag, such as "Second follow-up visit" and "High-intent customer"). For example, one record has the values c_id:1, staff_id:002, status:Ongoing follow-up, and tag:Second follow-up visit. Another record has the values c_id:1, staff_id:001, status:Follow-up, and tag:High-intent customer.

  • Destination Logstore: client-information

• ApsaraDB RDS resources

  • Database: client-db

  • Database table: client

    The client table contains five fields: c_id, name, telephone, update_time, and delete_flag. Three sample records are: c_id=1, name=maki, telephone=010-123; c_id=2, name=evan, telephone=010-156; and c_id=3, name=vicky, telephone=010-166. The update_time for all records is 1606358931, and delete_flag is false.

  • Database username and password: test/test1234@@

  • Public endpoint of the database: rm-bp1k****tp8o.mysql.rds.aliyuncs.com

Procedure

1. Log on to the Simple Log Service console.

2. Go to the data transformation page.

   1. In the Projects section, click client-project.

   2. On the Log Storage > Logstores tab, click the source Logstore client-log.

   3. On the query and analysis page, click Data Transformation.

3. In the upper-right corner of the page, select a time range.

   Verify that log data exists on the Raw Logs tab.

4. In the editor, enter the data transformation statement.

   For more information about the parameters, see res_rds_mysql.

   e_table_map(
       res_rds_mysql(
           "rm-bp1k****tp8o.mysql.rds.aliyuncs.com",     
           "test",
           "test1234@@",
           "client-db",
           table="client",
           fields=["c_id", "name", "telephone", "update_time"],
           refresh_interval=1,
           primary_keys="c_id",
           update_time_key="update_time",
           deleted_flag_key=None,
       ),
       "c_id",
       ["name", "telephone"],
   )

5. Preview the data in quick mode.

   Use the quick preview feature to verify that the data transformation statement is correct. For more information, see Quick preview.

   1. Click Quick.

   2. On the Test Data > Data tab, enter the following content.

      {
        "__source__": "192.0.2.0",
        "__time__": 1624956516,
        "__topic__": "log",
        "__tag__:__client_ip__":"192.0.2.2",
        "c_id": "1",
        "staff_id": "002",
        "status": "Ongoing follow-up",
        "tag": "Second follow-up visit",
      }

   3. On the Test Data > Dimension Table tab, enter the following content.

      c_id,name,telephone,update_time,delete_flag
      1,maki,010-123,1606358931,false

   4. Click Preview Data.

      View the preview results. The left side of the page displays the transformed data, with the output destination as target0. The transformed fields include customer information fields such as __source__, __tag__, __topic__, c_id, name, staff_id, status, tag, and telephone. On the right, the Run Result Summary shows Total: 1, Success: 1, and Failed: 0. This indicates that the data transformation rule ran correctly.

6. Preview the data in advanced mode.

   Use the advanced preview feature to ensure that Simple Log Service can connect to the ApsaraDB RDS for MySQL database. For more information, see Advanced preview.

   1. Click Advanced.

   2. Click Preview Data.

   3. In the Add Preview Settings panel, configure the authorization method and click OK.

      The authorization methods include default role, custom role, and access key. If you select default role, the role ARN defaults to acs:ram::{AccountID}:role/aliyunlogetlrole.

   4. View the preview results.

      The Run Result Summary panel on the right displays four statistics: Total Raw Logs, Dropped Count, Dispatched Count, and Failed. Confirm that the dispatched count matches the total raw logs and the failed count is 0. This indicates that the transformation rule ran correctly. The log output list on the left can be expanded to view details of the transformed sample log, which includes key-value pairs such as __source__, __topic__, name, and status.

      If a runtime error occurs, see Troubleshoot ApsaraDB RDS for MySQL data retrieval errors.

7. Create a data transformation job.

   1. Click Save as Transformation Job.

   2. In the Create Data Transformation Job panel, configure the parameters and click OK.

      For more information about the parameters, see Create a data transformation job. The configuration items include Rule Name (for example, rds-sls), Authorization Method (options are default role, custom role, or access key), and Role ARN. In the Storage Destination section, configure Destination Name (for example, rds), Destination Region (for example, China (Hangzhou)), Destination Project (for example, client-project), Destination Logstore (for example, client-information), and the corresponding Authorization Method and Role ARN. To deliver data to multiple destinations, click Add.

      After you create a data transformation job, you can view the transformed logs in the destination Logstore. After the job runs successfully, the destination Logstore contains transformed log records, including structured fields and their values, such as c_id, name, staff_id, status, tag, and telephone.

FAQ

How do I use the delete feature in incremental pulling mode?

In incremental pulling mode, Simple Log Service pulls only new or updated data based on the primary key and timestamp fields of the database table. If you mark a record for deletion in the database table (for example, by setting delete_flag=true), Simple Log Service still pulls that record for processing. To handle this scenario, the res_rds_mysql function includes the deleted_flag_key parameter. After you configure the deleted_flag_key parameter, the data transformation job fetches updates from the database and then removes rows from its in-memory dimension table where the delete_flag is true. This operation does not affect the database table. The job no longer includes these deleted rows in subsequent JOIN operations with log data.

For example, two new rows (name=mia and name=tom) are added to the ApsaraDB RDS for MySQL table. The row where name=mia is marked as deleted by setting delete_flag to true. When the in-memory dimension table of Simple Log Service is updated, the row for name=mia is deleted from it and is not processed.

The following code provides an example of the data transformation statement:

e_table_map(
    res_rds_mysql(
        "rm-bp1****l3tp.mysql.rds.aliyuncs.com",     
        "test",
        "test1234@@",
        "client-db",
        table="client",
        fields=["c_id", "name", "telephone", "update_time"],
        refresh_interval=1,
        primary_keys="c_id",
        update_time_key="update_time",
        deleted_flag_key="delete_flag",
    ),
    "c_id",
    ["name", "telephone"],
)