DataWorks:Synchronize data

Go to DataStudio

Log on to the DataWorks console. In the target region, click Data Development and O&M > Data Development in the left-side navigation pane. Select a workspace from the drop-down list and click Go to Data Development.

Design the workflow

1. Create a workflow.

   In DataWorks, you perform data development within workflows using development components. You must create a workflow before you create a node. For more information, see Create a workflow.

   Name the workflow user_profile_analysis_MaxCompute.

2. Design the workflow.

   After you create the workflow, its canvas automatically opens. Based on the workflow design, click New Node on the canvas. Drag the required nodes to the canvas and draw lines between them to set dependencies for the data synchronization stage.

   

3. In this tutorial, there is no data lineage between the zero load node and the batch synchronization nodes. Therefore, you must define their dependencies by drawing lines in the workflow. For more information about other ways to configure dependencies, see Configure scheduling dependencies. The following table describes the category, type, name, and function of each node.

   Category	Type	Node name (Named after the final output table)	Description
   General	zero load node	workshop_start_odps	Manages the entire user profile analysis workflow, such as setting when the workflow is triggered. When a workflow is complex, this node helps clarify the data flow path. This is a dry-run node and does not require 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 the scheduling logic

The workshop_start_odps zero load node triggers the workflow to run at 00:30 every day. The following table describes its key scheduling configurations. You do not need to change the scheduling settings for other nodes. For more information about the implementation logic, see Advanced time property configuration. For more information about other scheduling configurations, see Overview of task scheduling properties.

Create the destination MaxCompute tables

Create MaxCompute tables to store the raw data synchronized by Data Integration. For more information, see Create and use MaxCompute tables.

1. Create a table.

   In the left-side navigation pane, right-click the Table node under MaxCompute and select Create Table from the shortcut menu.

2. Create the ods_raw_log_d_odps table.

   In the Create Table dialog box, enter the name ods_raw_log_d_odps. On the table editor page, click DDL, enter the following DDL statement, click Generate Table Schema, and then click Confirm to overwrite the current configuration.

   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 the name ods_user_info_d_odps. On the table editor page, click DDL, enter the following DDL statement, click Generate Table Schema, and then click Confirm to overwrite the current configuration.

   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 sign'
   )
   PARTITIONED BY (
    dt STRING
   )
   LIFECYCLE 7;

4. Commit and deploy the tables.

   After you confirm the table information, click Commit to Development Environment and Commit to Production Environment for both the ods_user_info_d_odps and ods_raw_log_d_odps tables. This action creates the physical tables in the corresponding computing engine projects for the development and production environments based on your node configuration.

   Note

   You can view the table in the computing engine project of the corresponding environment only after a successful commit.

   • Committing a table to the DataWorks development environment creates the table in the MaxCompute computing engine bound to the development environment.

   • Committing a table to the DataWorks production environment creates the table in the MaxCompute computing engine bound to the production environment.

Create the source data sources

This tutorial uses test data from a MySQL database and Object Storage Service (OSS). 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 test data.

Configure the user data synchronization pipeline

Configure a batch synchronization node to synchronize basic user information from the MySQL table ods_user_info_d to the MaxCompute table ods_user_info_d_odps.

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

2. Configure network and resources.

   After you configure the Data source, My Resource Group, and Data going, click The next Step and complete the connectivity test as prompted. The following table describes the required configuration.

   Parameter	Configuration
   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 the environment step.
   Data going	Data going: MaxCompute Data Source Name: user_behavior_analysis_odps

3. Configure the task.

   • Configure the source and destination.

     Module	Parameter	Configuration
     Data source	Table	Select the MySQL table ods_user_info_d.
     	Shard Key	Use the primary key or an indexed column as the split key. Only integer-type fields are supported. For this tutorial, set the split key to the uid field.
     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 more information about MaxCompute data transmission resources, see Purchase and use exclusive resource groups for data transmission. If an exclusive Tunnel quota is unavailable due to an overdue payment or expiration, the task automatically switches to use common transmission resources.
     	Schema	This tutorial uses the default schema. If you have other schemas in your MaxCompute project, you can select one from the drop-down list.
     	Table	Select the ods_user_info_d_odps table that you created earlier.
     	Partition Information	For this tutorial, enter the ${bizdate} parameter.
     	Write Method	For this tutorial, select Clean up existing data before writing (Insert Overwrite). The following write modes are supported: insert into: Inserts data directly into a table or static partition. insert overwrite: Clears the existing data in a table or static partition before inserting new data.
     	Write by Converting Empty Strings into Null	For this tutorial, select No.

   • Confirm field mapping and general settings.

     DataWorks lets you map source fields to destination fields. This ensures data from specified source fields is written to the correct destination fields. It also provides features for setting concurrent read/write operations, rate limiting to prevent synchronization from affecting database performance, defining dirty data handling, and executing distributed tasks. This tutorial uses the default settings. For more information about other configuration options, see Configure a batch synchronization node by using the codeless UI.

4. Configure scheduling properties.

   On the configuration page, click Scheduling settings in the right-side pane to open the Scheduling Configuration panel. Here, you can configure scheduling and node information. For details, see Configure scheduling properties for a node. For this tutorial, use the following configuration:

   Parameter	Configuration
   Scheduling Parameters	Use the default value $bizdate. Note bizdate=$bizdate retrieves the date of the previous day in the yyyymmdd format.
   Time attribute	Scheduling period: Day. Scheduling Time: 00:30. RUN Attribute: Allows the task to be rerun regardless of its previous success or failure. Use the default values for other parameters. Note The node runs only after 00:30 every day.
   Resource Group	Select the Serverless resource group that you created in the Prepare the environment step.
   Scheduling Dependency	Confirm the Parent Nodes: Confirm that the workshop_start node is displayed as a parent node of the current node. The parent dependencies that you set by connecting nodes are displayed here. If the workshop_start node is not displayed, make sure that you have designed the business flow for the data synchronization stage by referring to Design a business flow. In this use case, the current node is triggered after the workshop_start node runs at its scheduled time and completes successfully. Confirm that a node output named [Name of the MaxCompute production project].ods_user_info_d_odps exists. If not, you must manually add it with this output name. Note DataWorks uses node outputs to manage dependencies. The automatic parsing feature uses table lineage to help downstream SQL nodes process the output of a synchronization task by automatically adding the required dependency. You must confirm whether a node output exists with the same name as the synchronization output table, [Name of the MaxCompute production project].ods_user_info_d_odps.

Configure the user log synchronization pipeline

Configure a batch synchronization node to synchronize user website access data from the user_log.txt file in the public HttpFile data source to the MaxCompute table ods_raw_log_d_odps.

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

2. Configure network and resources.

   After you configure the Data source, My Resource Group, and Data going, click The next Step and complete the connectivity test as prompted. The following table describes the required configuration.

   Parameter	Configuration
   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 the environment step.
   Data going	Data going: MaxCompute Data Source Name: user_behavior_analysis_odps

3. Configure the task.

   • Configure the source and destination.

     Module	Parameter		Configuration
     Data source	File Path		For this tutorial, enter /user_log.txt.
     	File Type		Select text.
     	Field Delimiter		For this tutorial, enter |.
     	Advanced Configuration	Encoding	Select the UTF-8 encoding format.
     		Compression Format	Select None.
     		Skip Header	Select No.
     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 more information about MaxCompute data transmission resources, see Purchase and use exclusive resource groups for data transmission. If an exclusive Tunnel quota is unavailable due to an overdue payment or expiration, the task automatically switches to use common transmission resources.
     	Schema		This tutorial uses the default schema. If you have other schemas in your DataWorks project, you can select one from the drop-down list.
     	Table		Select the ods_raw_log_d_odps table that you created earlier.
     	Partition Information		For this tutorial, enter the ${bizdate} parameter.
     	Write Method		For this tutorial, select Clean up existing data before writing (Insert Overwrite). The following write modes are supported: insert into: Inserts data directly into a table or static partition. insert overwrite: Clears the existing data in 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 the log file can be read correctly.

   • Confirm field mapping and general settings.

     DataWorks lets you map source fields to destination fields. This ensures data from specified source fields is written to the correct destination fields. It also provides features for setting concurrent read/write operations, rate limiting to prevent synchronization from affecting database performance, defining dirty data handling, and executing distributed tasks. This tutorial uses the default settings. For more information about other configuration options, see Configure a batch synchronization node by using the codeless UI.

4. Configure scheduling properties.

   On the configuration page, click Scheduling settings in the right-side pane to open the Scheduling Configuration panel. Here, you can configure scheduling and node information. For details, see Configure scheduling properties for a node. For this tutorial, use the following configuration:

   Parameter	Configuration
   Scheduling Parameters	Use the default value $bizdate. Note bizdate=$bizdate retrieves the date of the previous day in the yyyymmdd format.
   Time attribute	Scheduling period: Day. Scheduling Time: 00:30. RUN Attribute: Allows the task to be rerun regardless of its previous success or failure. Use the default values for other parameters. Note The node runs only after 00:30 every day.
   Resource Group	Select the Serverless resource group that you created in the Prepare the environment step.
   Scheduling Dependency	Confirm the Parent Nodes: Check that the workshop_start node is displayed as a parent node of the current node. The parent nodes that you set by connecting nodes are displayed here. If the workshop_start node is not displayed, make sure that you have designed the business flow for the business data synchronization phase as described in Design a business flow. In this use case, the current node is triggered after the workshop_start node runs at its scheduled time and completes successfully. Confirm that a node output named [Name of the MaxCompute production project].ods_raw_log_d_odps exists. If not, you must manually add it with this output name. Note DataWorks uses node outputs to manage dependencies. The automatic parsing feature uses table lineage to help downstream SQL nodes process the output of a synchronization task by automatically adding the required dependency. You must confirm whether a node output exists with the same name as the synchronization output table, [Name of the MaxCompute production project].ods_raw_log_d_odps.

Run the business flow

1. On the DataStudio page, double-click User Profile Analysis_MaxCompute under Workflow. After the Business Flow panel opens, click the icon in the toolbar to run the Business Flow based on its upstream and downstream dependencies.

2. Verify the execution:

   • Check the task running status: A node in the state indicates that the synchronization process completed successfully.

   • View the run logs: Right-click the ods_user_info_d_odps and ods_raw_log_d_odps nodes and select View Logs. The presence of the following keywords in the logs indicates that the node ran successfully and synchronized the data.

     Keywords indicating a successful run include completed successfully, Shell run successfully!, and Exit with SUCCESS.

Verify the synchronization results

After the Business Flow runs, it synchronizes data from the MySQL ods_user_info_d table to the previous day's partition in workshop2024_01_dev.ods_user_info_d_odps, and logs from the OSS user_log.txt file to the previous day's partition in workshop2024_01_dev.ods_raw_log_d_odps. Since the SQL query does not need to be published to the production environment, you can create an ad hoc query file to verify the results.

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 destination tables.

   Run the following SQL statements to confirm that the data was written correctly. Check the number of records imported into the ods_raw_log_d_odps and ods_user_info_d_odps tables.

   // Replace 'your_business_date' with the actual business date. For example, if the task ran on 20230621, the business date is 20230620 (the day before the run date).
   select count(*) from ods_user_info_d_odps  where dt='your_business_date'; 
   select count(*) from ods_raw_log_d_odps where dt='your_business_date';

   If the synchronization is successful, ods_user_info_d_odps should return a non-zero count (for example, 6), and ods_raw_log_d_odps should also return a non-zero count, indicating the data was successfully written to the corresponding partitions.

   Note

   Since this tutorial runs in DataStudio (development environment), data is written by default to the specified tables in the MaxCompute project workshop2024_01_dev for the development environment.