DataWorks:Synchronize data

DataStudio

Log on to the DataWorks console. In the top navigation bar, select the desired region. In the left-side navigation pane, choose Data Development and O&M > Data Development. On the page that appears, select the desired workspace from the drop-down list and click Go to Data Development.

Design a workflow

1. Create a workflow.

   In DataWorks, you perform data development using components within a workflow. Before you create a node, you must create a workflow. For instructions, see Create a workflow.

   Name the workflow user_profile_analysis_MaxCompute.

   

2. Design the workflow.

   After you create the workflow, its canvas opens automatically. On the canvas, click New Node. Design the data synchronization workflow by dragging nodes to the canvas and connecting them to define dependencies. For more information, see Workflow design.

   

3. In this tutorial, no data lineage exists between the zero-load node and the batch synchronization nodes. Therefore, define their dependencies by drawing lines on the workflow canvas. For more information about how to configure dependencies, see Guide for configuring scheduling dependencies. The table below describes the type, name, and function of each node.

   Category	Type	Name (Named after the final output table)	Description
   General	zero-load node	workshop_start_odps	Orchestrates the user profile analysis workflow, for example, by setting its start time. It clarifies the data flow path in complex workflows. This node is a dry-run task and requires no code editing.
   Data integration	batch synchronization	ods_user_info_d_odps	Synchronizes basic user information from MySQL to the MaxCompute table ods_user_info_d_odps.
   Data integration	batch synchronization	ods_raw_log_d_odps	Synchronizes user website access logs from OSS to the MaxCompute table ods_raw_log_d_odps.

Configure scheduling logic

In this tutorial, the workshop_start_odps zero-load node triggers the workflow to run at 00:30 every day. The table below details the key scheduling configurations for this node. You do not need to change the scheduling settings for other nodes. For implementation details, see Advanced time property configurations. For more information about other scheduling configurations, see Overview of task scheduling property configurations.

Create MaxCompute tables

Before you start, create the destination MaxCompute tables that will store the raw data that Data Integration synchronizes. This tutorial explains how to quickly create them. For more detailed instructions, see Manage MaxCompute tables.

1. Navigate to the page for creating tables.

   

2. Create the ods_raw_log_d_odps table.

   In the Create Table dialog box, enter ods_raw_log_d_odps in the Name field. On the table editor page, click DDL, enter the following statement, and then click Generate Table Schema. Click Confirm to overwrite the current schema.

   CREATE TABLE IF NOT EXISTS ods_raw_log_d_odps
   (
    col STRING
   ) 
   PARTITIONED BY
   (
    dt STRING
   )
   LIFECYCLE 7;

3. Create the ods_user_info_d_odps table.

   In the Create Table dialog box, enter ods_user_info_d_odps in the Name field. On the table editor page, click DDL, enter the following statement, and then click Generate Table Schema. Click Confirm to overwrite the current schema.

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

4. Commit and publish the tables.

   After you verify that the table information is correct, for each of the ods_user_info_d_odps and ods_raw_log_d_odps tables, click Commit to Development Environment and then Commit to Production Environment. This operation creates physical tables in the compute engine projects for the development and production environments, as defined in your node configuration.

   Note

   After you define a table schema, you must commit the table to both the development and production environments. After the table is successfully committed, you can view it in the corresponding compute engine project.

   • Commit the table to the DataWorks development environment to create the table in the MaxCompute engine associated with the development environment.

   • Commit the table to the DataWorks production environment to create the table in the MaxCompute engine associated with the production environment.

Create data sources

This tutorial uses test data from an ApsaraDB for RDS (MySQL) instance and OSS. You must create a MySQL data source named user_behavior_analysis_mysql and an HttpFile data source named user_behavior_analysis_httpfile in your workspace to access the provided test data. The basic information required for the data sources is provided.

Configure user data synchronization

This batch synchronization node synchronizes basic user information data from the MySQL table ods_user_info_d provided in the use case to the MaxCompute table ods_user_info_d_odps.

1. Double-click the batch synchronization node ods_user_info_d_odps to open the node editing page.

2. Network and resource configuration.

   Configure the Data source, My Resource Group, and Data going, and then click The next Step. Perform the connectivity test as prompted. The following table describes these settings.

   

   Parameter	Description
   Data source	Data source: MySQL Data Source Name: user_behavior_analysis_mysql
   My Resource Group	Select the serverless resource group that you purchased in the Prepare environment step.
   Data going	Data going: MaxCompute Data Source Name: user_behavior_analysis_odps

3. Task configuration.

   • Configure the source and destination.

     Module	Parameter	Description	Illustration
     Data source	Table	Select the MySQL table ods_user_info_d.
     	Shard Key	Use a primary key or an indexed column as the split key. The split key must be an integer field. Here, the split key is set to the uid field.
     Data going	Tunnel Resource Group	This tutorial uses the default Common transmission resources. If you have an exclusive tunnel quota, you can select it from the drop-down list. Note For more information about MaxCompute data transmission resources, see Purchase and use exclusive data transfer resource groups. If your exclusive tunnel quota is unavailable due to an overdue payment or expiration, the task automatically switches to common transmission resources during runtime.
     	schema	In this tutorial, select default. If your MaxCompute project has other schemas, you can select one from the drop-down list.
     	Table	Select the ods_user_info_d_odps table that is created in the ad hoc query.
     	Partition Information	This tutorial uses the ${bizdate} parameter.
     	Write Method	For this tutorial, select "Clean up existing data before writing" from the drop-down list. Two write modes are available: insert into: Directly inserts data into a table or static partition. insert overwrite: Clears existing data from a table or static partition before inserting new data.
     	Write by Converting Empty Strings into Null	For this tutorial, select No.

   • Confirm field mappings and general settings.

     DataWorks lets you map fields between the source and destination. You can also configure settings for read/write concurrency, rate limits to avoid impacting the database, dirty data handling, and distributed task execution. This tutorial uses the default settings. For more information about other configuration options, see Wizard Mode Configuration.

4. Scheduling Properties.

   On the node configuration page, click Properties in the right-side navigation pane to open the Scheduling Configuration panel. On this panel, configure scheduling and node information. For more information, see Node Scheduling Configuration. The following table describes the configurations.

   Parameter	Description	Illustration
   Scheduling Parameters	You can leave the scheduling parameter value as the default $bizdate. Note bizdate=$bizdate returns the previous day's date in yyyymmdd format.
   Time attribute	Scheduling period: Day. Scheduling Time: 00:30. RUN Attribute: Allow Regardless of Running Status. Keep the default values for other parameters. Note This node's daily start time depends on the scheduled time of the workshop_start virtual node. The node runs only after 00:30 each day.
   Resource Group	Select the serverless resource group that you created in the Prepare environment step.
   Scheduling Dependency	Confirm the Parent Nodes: Confirm that the workshop_start node is displayed as the upstream node of the current node. Upstream dependencies that you configure by drawing lines are displayed here. If the workshop_start node is not displayed, make sure that you have completed the business flow design for the data synchronization stage as described in Design the business flow. In this example, the current node will be triggered to run after the scheduled time for the workshop_start node is reached and the node completes its execution. Confirm this node's output: check whether a node output named MaxCompute_production_project_name.ods_user_info_d_odps exists. If this node output does not exist, manually add the Node Output. Note DataWorks establishes node dependencies by using node outputs. To make it easier for a downstream SQL task to process the output table of a synchronization task, you can use the automatic parsing mechanism to quickly add the dependency based on table lineage. In this case, you must confirm that a node output exists with the same name as the synchronization output table, such as MaxCompute_production_project_name.ods_user_info_d_odps.

Configure the user log pipeline

This batch synchronization node is used to configure a synchronization task to synchronize user website access information from the public HttpFile object user_log.txt to the MaxCompute table ods_raw_log_d_odps.

1. Double-click the batch synchronization node ods_raw_log_d_odps to open the node editing page.

2. Network and resource configuration.

   Configure the Data source, My Resource Group, and Data going. Then, click The next Step and run the connectivity test as prompted. The following table describes the configurations.

   

   Parameter	Description
   Data source	Data source: HttpFile Data Source Name: user_behavior_analysis_HttpFile
   My Resource Group	Select the serverless resource group that you purchased in the Prepare environment step.
   Data going	Data going: MaxCompute Data Source Name: user_behavior_analysis_odps

3. Task configuration.

   • Configure the source and destination.

     Module	Parameter		Description	Illustration
     Data source	File Path		In this tutorial, enter /user_log.txt.
     	File Type		Select the text type.
     	Field Delimiter		In this tutorial, enter: |.
     	Advanced Configuration	Encoding	Select the UTF-8 encoding format.
     		Compression Format	Select the None format.
     		Skip Header	Select No to not skip the table header.
     Data going	Tunnel Resource Group		This tutorial uses common transmission resources by default. If you have an exclusive Tunnel quota, you can select it from the drop-down list. Note For information about selecting data transmission resources for MaxCompute, see Purchase and use exclusive data transmission service resource groups. If an exclusive Tunnel quota is unavailable due to overdue payments or expiration, the task automatically switches to common transmission resources while it is running.
     	Schema		For this tutorial, select the default schema. If your DataWorks project has other schemas, you can select one from the drop-down list.
     	Table		From the drop-down list, select the ods_raw_log_d_odps table created in the ad-hoc query.
     	Partition Information		This tutorial uses the ${bizdate} parameter.
     	Write Method		For this tutorial, select Clean up existing data before writing. The following write modes are supported: insert into: Directly inserts data into a table or static partition. insert overwrite: Clears existing data from a table or static partition before inserting new data.
     	Write by Converting Empty Strings into Null		For this tutorial, select No.

     After you configure the source, click Confirm Data Structure to verify that DataWorks can read the log file correctly.

   • Confirm field mapping and general settings.

     DataWorks allows you to map source fields to destination fields. It also provides features such as parallelism, rate limiting (to avoid impacting your database), dirty data handling, and distributed task execution. For this tutorial, use the default settings. For more information about other configuration options, see Configure a batch synchronization task in wizard mode.

4. Configure scheduling properties.

   On the configuration page, click Scheduling configuration in the right-side panel. Then, in the Scheduling Configuration panel, configure scheduling and node information. For more information, see Configure scheduling properties for a node. The following table describes the configurations.

   Parameter	Description	Illustration
   Scheduling Parameters	You can leave the scheduling parameter value as the default $bizdate. Note bizdate=$bizdate returns the previous day's date in yyyymmdd format.
   Time attribute	Scheduling period: Day. Scheduling Time: 00:30. RUN Attribute: Allow Regardless of Running Status. Keep the default values for other parameters. Note This node's daily start time depends on the scheduled time of the workshop_start virtual node. The node runs only after 00:30 each day.
   Resource Group	Select the serverless resource group you created in the Prepare environment stage.
   Scheduling Dependency	Confirm the Parent Nodes: Verify that the workshop_start node is displayed as an upstream node of the current node. Upstream dependencies that are configured by connecting nodes are displayed in this section. If the workshop_start node is not displayed, make sure that you have completed the business flow design for the business data synchronization stage as described in Design a business flow. In this example, when the scheduled time for the workshop_start node is reached and its execution is complete, the current node is triggered to run. Confirm the node output: Check whether a node output named YourMaxComputeProjectName.ods_raw_log_d_odps exists. If the node output does not exist, you must manually add the Node Output. Note DataWorks uses node outputs to establish node dependencies. When a downstream SQL task processes the output table of a synchronization task, the automatic parsing mechanism uses table lineage to quickly add the synchronization task as a dependency for the SQL node. For this mechanism to work, you must confirm that a node output exists with the same name as the synchronization output table: MaxCompute production project name.ods_raw_log_d_odps.

Run the business flow

1. On the DataStudio page, double-click user_profile_analysis_MaxCompute under Workflow to open the Workflow panel, and then click the icon in the toolbar to run the workflow according to its upstream and downstream dependencies.

2. Check the execution status:

   • Check the node status: A status indicates that the synchronous execution process has no issues.

   • To view the task execution log, right-click the ods_user_info_d_odps or ods_raw_log_d_odps node and select View Logs. When the following text appears in the log, it indicates that the synchronization node ran successfully and the data was synchronized successfully.

     

View the synchronization results

Running the business workflow is expected to fully synchronize the data from the basic user information table ods_user_info_d in MySQL to yesterday's partition of the workshop2024_01_dev.ods_user_info_d_odps table, and the user website access logs from the user_log.txt file in OSS to yesterday's partition of the workshop2024_01_dev.ods_raw_log_d_odps table. Because the query SQL does not need to be run in the production environment, you can create an ad hoc query file to verify the result.

1. Create an ad hoc query file.

   In the left-side navigation pane of the DataStudio page, click the icon to open the Ad Hoc Query pane. Right-click Ad Hoc Query and select New Node > ODPS SQL.

2. Query the synchronization result tables.

   Execute the following SQL statement to verify the result of the data synchronization. Check the number of records imported into ods_raw_log_d_odps and ods_user_info_d_odps.

   // Replace 'your_data_timestamp' with the actual data timestamp, which is the day before the task's run date (e.g., use 20230620 if the task runs on 20230621).
   select count(*) from ods_user_info_d_odps  where dt='your_data_timestamp'; 
   select count(*) from ods_raw_log_d_odps where dt='your_data_timestamp';

   

   Note

   In this tutorial, you run nodes in DataStudio, which is the development environment. Therefore, data is written by default to tables in the workshop2024_01_dev project, which is the compute engine project for this environment.