DataWorks:Synchronize data

Create a MySQL data source (user_behavior_analysis_mysql)

The basic user information for this tutorial is stored in a MySQL database. You need to create a MySQL data source to synchronize the user data ( ods_user_info_d) to the private OSS bucket that you created when you prepared the environment.

1. Go to the Data Source page.

   1. Log on to the DataWorks console. In the top navigation bar, select the desired region. In the left-side navigation pane, choose More > Management Center. On the page that appears, select the desired workspace from the drop-down list and click Go to Management Center.

   2. In the navigation pane on the left, click Data Source to go to the Data Source page.

2. Click Add Data Source. Search for and select MySQL.

3. On the Create MySQL Data Source page, configure the parameters. For this tutorial, use the same example values for both the development and production environments.

   The following table describes the key parameters. You can keep the default values for the other parameters.

   Parameter	Description
   Data Source Name	Enter a name for the data source. This tutorial uses user_behavior_analysis_mysql.
   Data Source Description	This data source is for DataWorks tutorials. Read from this data source in a batch synchronization configuration to access the test data provided by the platform. This data source can be read from only in Data Integration scenarios. Other modules cannot use it.
   Configuration Mode	Select Connection String Mode.
   Connection String	Host IP Address: rm-bp1z69dodhh85z9qa.mysql.rds.aliyuncs.com Port: 3306
   Database Name	Enter the database name. This tutorial uses workshop.
   Username	Enter the username. This tutorial uses workshop.
   Password	Enter the password. This tutorial uses workshop#2017.
   Authentication Option	No authentication.

4. In the Connection Configuration section, click Test Connectivity for the development and production environments. Ensure that the status is Connected.

   Important

   • Ensure that the resource group is attached to the workspace and has Internet access. Otherwise, an error will occur during data synchronization. For more information, see Prepare an environment.

   • If no resource groups are available, follow the prompts in the connection configuration section. Click Purchase or Attach Purchased Resource Group.

5. Click Complete.

Create an HttpFile data source named user_behavior_analysis_httpfile

The user website access records for this tutorial are stored in OSS. You need to create an HttpFile data source to synchronize the user website access records ( user_log.txt) to the private OSS bucket that you created when you prepared the environment.

1. On the Management Center page, click Data Source in the navigation pane on the left.

2. Click Add Data Source. In the Add Data Source dialog box, search for and select HttpFile.

3. On the Create HttpFile Data Source page, configure the parameters. For this tutorial, use the same example values for both the development and production environments.

   The following table describes the key parameters. You can keep the default values for the other parameters.

   Parameter	Description
   Data Source Name	Enter a name for the data source. This tutorial uses user_behavior_analysis_httpfile.
   Data Source Description	This data source is for DataWorks tutorials. Read from this data source in a batch synchronization configuration to access the test data provided by the platform. This data source can be read from only in Data Integration scenarios. Other modules cannot use it.
   URL Domain Name	Set the URL Domain Name for both the development and production environments to https://dataworks-workshop-2024.oss-cn-shanghai.aliyuncs.com.

4. In the Connection Configuration section, click Test Connectivity for the development and production environments. Ensure that the status is Connected.

   Important

   • Ensure that the resource group is attached to the workspace and has Internet access. Otherwise, an error will occur during data synchronization. For more information, see Prepare an environment.

   • If no resource groups are available, follow the prompts in the connection configuration section. Click Purchase or Attach Purchased Resource Group.

5. Click Complete.

Create an OSS data source

In this tutorial, the user information from the MySQL data source and the log information from the HttpFile data source are synchronized to an OSS data source.

1. Go to the Data Sources page and click Add Data Source.

2. In the Add Data Source dialog box, search for and select OSS.

3. On the Add OSS Data Source page, configure the parameters. For this tutorial, use the example values for both the development and production environments.

   Parameter	Description
   Data Source Name	Enter a name for the data source. In this example, enter test_g.
   Data Source Description	Enter a brief description for the data source.
   Access Mode	Select Access Key Mode.
   AccessKey ID	The AccessKey ID of the current logon account. You can go to the AccessKey page to copy the AccessKey ID.
   AccessKey Secret	Enter the AccessKey secret of the current logon account. Important The AccessKey secret is displayed only when you create it. You cannot view it later. Keep your AccessKey secret confidential. If an AccessKey is leaked or lost, delete it and create a new one.
   Endpoint	Enter http://oss-cn-shanghai-internal.aliyuncs.com.
   Bucket	The name of the private OSS bucket that you prepared. In this example, the name is dw-spark-demo.

4. For the specified resource group, click Test Network Connectivity in the Connection Status (Development Environment) and Connection Status (Production Environment) columns. Wait for the test to complete and ensure that the status is Connected.

   Note

   Ensure that at least one resource group is in the Connected state. Otherwise, you cannot use the codeless UI to create a sync task for this data source.

5. Click Complete Creation to create the OSS data source.

Configure the initial node

1. On the workflow orchestration page, hover over the workshop_start_spark node and click Open Node.

2. In the right-side pane of the workshop_start_spark node editing page, click Scheduling and configure the parameters. The following table describes the key parameters. You can keep the default values for any parameters that are not mentioned.

   Scheduling configuration parameter	Description
   Scheduling Type	For this tutorial, set the type to Dry-run.
   Resource Group	For this tutorial, select the Serverless resource group created when you prepared the environment.
   Scheduling Dependencies	Because workshop_start_spark is the initial node, it has no upstream dependencies. Click Use Workspace Root Node to have the workspace root node trigger the workflow. The workspace root node is named: WorkspaceName_root.

Configure the user log synchronization pipeline (ods_raw_log_d_2oss_spark)

The ods_raw_log_d_2oss_spark batch synchronization node synchronizes user log information from the HttpFile data source to the private OSS data source.

1. On the workflow orchestration page, hover over the ods_raw_log_d_2oss_spark node and click the open node button to go to the node configuration page.

2. Configure the network and resources.

   Parameter	Description
   Source	Source: HttpFile. Data Source Name: user_behavior_analysis_httpfile.
   Resource Group	Select the Serverless resource group purchased during environment preparation.
   Destination	Destination: OSS. Data Source Name: Select the private OSS data source created earlier. In this example, select test_g.

3. Click Next to configure the sync task.

   • Configure Source and Destination

     The following table describes the key parameters. You can keep the default values for any parameters that are not mentioned.

     Parameter	Description
     Source	File Path: /user_log.txt. Text Type: Select text. Column Delimiter: Enter |. Compression Format: Includes None, Gzip, Bzip2, and Zip. Select None. Skip Header: Select No.
     Destination	Text type: Select text. File name (including path): Enter a path based on the directory in your OSS bucket. Example: ods_raw_log_d/log_${bizdate}/log_${bizdate}.txt. Here, ods_raw_log_d is the directory name you created, and $bizdate represents the date of the previous day. Column Delimiter: Enter |.

   • Confirm the Field Mapping and Channel Control settings.

     DataWorks maps source fields to destination fields. You can also configure the concurrency and the policy for handling dirty data. For this tutorial, set the Policy For Dirty Data parameter to Disallow Dirty Data. You can keep the default values for the other parameters. For more information, see Configure a batch synchronization node in the codeless UI.

4. Configure debugging parameters.

   In the right-side pane of the batch synchronization task configuration page, click Debugging Configurations. Configure the following parameters to run a test with debugging parameters in Step 6.

   Configuration item	Configuration description
   Resource Group	Select the serverless resource group purchased when you prepared the environment.
   Script Parameters	Click Add Parameter and assign a value to bizdate in yyyymmdd format, such as bizdate=20250223. During debugging, DataStudio will replace the variables defined in the task with this constant.

5. (Optional) Configure scheduling properties.

   You can keep the default values for the scheduling parameters in this tutorial. You can click Scheduling in the right-side pane of the node editing page. For more information about the parameters, see Configure scheduling properties for a node.

   • Scheduling Parameters: These are already configured for the workflow in this tutorial. You do not need to configure them for inner nodes. They can be used directly in tasks or code.

   • Scheduling Policies: You can set the Time for Delayed Execution parameter to specify how long a child node waits to run after the workflow starts. You do not need to set this parameter for this tutorial.

6. In the top toolbar, click Save to save the current node.

Configure the user data synchronization pipeline (ods_user_info_d_2oss_spark)

The ods_user_info_d_2oss_Spark batch synchronization node synchronizes user data from the MySQL data source to the private OSS data source.

1. On the workflow orchestration page, hover over the ods_user_info_d_2oss_Spark node and click Open Node.

2. Configure the synchronization pipeline network and resources.

   Parameter	Description
   Source	Data source: MySQL. Data source name: user_behavior_analysis_mysql.
   Resource Group	Select the serverless resource group purchased during environment preparation.
   Destination	Data destination: OSS. Data source name: Select the private OSS data source created earlier. In this example, select test_g.

3. Click Next to configure the sync task.

   • Configure Data Source and Destination

     The following table describes the key parameters. You can keep the default values for any parameters that are not mentioned.

     Parameter	Description
     Source	Table: Select ods_user_info_d from the data source. Split Key: Use a primary key or an indexed column as the shard key. Only integer fields are supported. Set the shard key to uid.
     Destination	Text Type: Select text. File Name (including Path): Enter a path based on the directory in your OSS bucket. Example: ods_user_info_d/user_${bizdate}/user_${bizdate}.txt. Here, ods_user_info_d is the directory name you created, and $bizdate represents the date of the previous day. Column Delimiter: Enter |.

   • Confirm the Field Mapping and Channel Control settings.

     DataWorks maps source fields to destination fields. You can also configure the concurrency and the policy for handling dirty data. For this tutorial, set the Policy For Dirty Data parameter to Disallow Dirty Data. You can keep the default values for the other parameters. For more information, see Configure a batch synchronization node in the codeless UI.

4. Configure debugging parameters.

   In the right-side pane of the batch synchronization task configuration page, click Debugging Configurations. Configure the following parameters to run a test with debugging parameters in Step 6.

   Configuration item	Configuration description
   Resource Group	Select the serverless resource group purchased when you prepared the environment.
   Script Parameters	Click Add Parameter and assign a value to bizdate in yyyymmdd format, such as bizdate=20250223. During debugging, DataStudio will replace the variables defined in the task with this constant.

5. (Optional) Configure scheduling properties.

   You can keep the default values for the scheduling parameters in this tutorial. You can click Scheduling in the right-side pane of the node editing page. For more information about the parameters, see Configure scheduling properties for a node.

   • Scheduling Parameters: These are already configured for the workflow in this tutorial. You do not need to configure them for inner nodes. They can be used directly in tasks or code.

   • Scheduling Policies: You can set the Time for Delayed Execution parameter to specify how long a child node waits to run after the workflow starts. You do not need to set this parameter for this tutorial.

6. In the top toolbar, click Save to save the current node.

Create a log table (ods_raw_log_d_spark) and parse data

After data is synchronized to the private OSS data source by the batch integration task, you can use an EMR SPARK SQL node to create the ods_raw_log_d_spark foreign table based on the generated OSS file. Use the LOCATION clause to access the log information that is written to the private OSS Bucket by the batch data integration task.

1. On the workflow orchestration page, hover over the ods_raw_log_d_spark node and click Open Node.

2. Edit the table creation statement.

   -- Scenario: The following Spark SQL code creates the ods_raw_log_d_spark foreign table using an EMR Spark SQL node. 
   -- It uses LOCATION to get the log information written to the private OSS bucket by the batch data integration task and adds the corresponding dt partition.
   -- Note:
   --      DataWorks provides scheduling parameters to write daily incremental data to the corresponding business partition of the destination table in a scheduling scenario.
   --      In a development scenario, you can define code variables in the ${VariableName} format. 
   --      Then, on the scheduling configuration page, you can assign scheduling parameters as values to the variables. This allows for dynamic parameter input in the code during scheduling. 
   CREATE EXTERNAL TABLE IF NOT EXISTS ods_raw_log_d_spark
   (
     `col` STRING
   ) 
   PARTITIONED BY (
     dt STRING
   )
   LOCATION 'oss://dw-spark-demo.oss-cn-shanghai-internal.aliyuncs.com/ods_raw_log_d/log_${bizdate}/';
   
   ALTER TABLE ods_raw_log_d_spark ADD IF NOT EXISTS PARTITION (dt = '${bizdate}') 
   LOCATION 'oss://dw-spark-demo.oss-cn-shanghai-internal.aliyuncs.com/ods_raw_log_d/log_${bizdate}/'
   

   Note

   Replace the location address in the code as needed. dw-spark-demo is the name of the OSS bucket that you created when you prepared the environment.

3. Configure debugging parameters.

   In the right-side pane of the EMR SPARK SQL task configuration page, click Debugging Configurations. Configure the following parameters to run a test with debugging parameters in Step 6.

   Configuration item	Configuration description
   Computing Resource	Select the Spark computing resource attached when you prepared the environment.
   Resource Group	Select the Serverless resource group purchased when you prepared the environment.
   Script Parameters	Click Add Parameter and assign a value to bizdate in yyyymmdd format, such as bizdate=20250223. During debugging, DataStudio will replace the variables defined in the task with this constant.

4. (Optional) Configure scheduling properties.

   You can keep the default values for the scheduling parameters in this tutorial. You can click Scheduling in the right-side pane of the node editing page. For more information about the parameters, see Configure scheduling properties for a node.

   • Scheduling Parameters: These are already configured for the workflow in this tutorial. You do not need to configure them for inner nodes. They can be used directly in tasks or code.

   • Scheduling Policies: You can set the Time for Delayed Execution parameter to specify how long a child node waits to run after the workflow starts. You do not need to set this parameter for this tutorial.

5. In the top toolbar, click Save to save the current node.

Create a user table (ods_user_info_d_spark) and parse data

After data is synchronized to the private OSS data source by the batch integration task, you can use an EMR SPARK SQL node to create the ods_user_info_d_spark foreign table based on the generated OSS file. Use the LOCATION clause to access the user information that is written to the private OSS Bucket by the batch data integration task.

1. On the workflow orchestration page, hover over the ods_user_info_d_spark node and click Open Node.

2. You can configure the network and resources for the synchronization link.

   -- Scenario: The following Spark SQL code creates the ods_user_info_d_spark foreign table using an EMR Spark SQL node. 
   -- It uses LOCATION to get the user information written to the private OSS bucket by the batch data integration task and writes it to the corresponding dt partition.
   -- Note:
   --      DataWorks provides scheduling parameters to write daily incremental data to the corresponding business partition of the destination table in a scheduling scenario.
   --      In a development scenario, you can define code variables in the ${VariableName} format. 
   --      Then, on the scheduling configuration page, you can assign scheduling parameters as values to the variables. This allows for dynamic parameter input in the code during scheduling.
   CREATE EXTERNAL TABLE IF NOT EXISTS ods_user_info_d_spark
   (
       `uid`        STRING COMMENT 'User ID'
       ,`gender`    STRING COMMENT 'Gender'
       ,`age_range` STRING COMMENT 'Age range'
       ,`zodiac`    STRING COMMENT 'Zodiac sign'
   )
   PARTITIONED BY 
   (
       dt           STRING
   )
   ROW FORMAT DELIMITED 
   FIELDS
   TERMINATED
   BY'|'
   STORED AS TEXTFILE
   LOCATION 'oss://dw-spark-demo.oss-cn-shanghai-internal.aliyuncs.com/ods_user_info_d/user_${bizdate}/'
   ;
   
   ALTER TABLE ods_user_info_d_spark ADD IF NOT EXISTS PARTITION (dt = '${bizdate}') 
   LOCATION'oss://dw-spark-demo.oss-cn-shanghai-internal.aliyuncs.com/ods_user_info_d/user_${bizdate}/'
   ;

   Note

   Replace the location address in the code as needed. dw-spark-demo is the name of the OSS bucket that you created when you prepared the environment.

3. Configure debugging parameters.

   In the right-side pane of the EMR SPARK SQL task configuration page, click Debugging Configurations. Configure the following parameters to run a test with debugging parameters in Step 6.

   Configuration item	Configuration description
   Computing Resource	Select the Spark computing resource attached when you prepared the environment.
   Resource Group	Select the Serverless resource group purchased when you prepared the environment.
   Script Parameters	Click Add Parameter and assign a value to bizdate in yyyymmdd format, such as bizdate=20250223. During debugging, DataStudio will replace the variables defined in the task with this constant.

4. (Optional) Configure scheduling properties.

   You can keep the default values for the scheduling parameters in this tutorial. You can click Scheduling in the right-side pane of the node editing page. For more information about the parameters, see Configure scheduling properties for a node.

   • Scheduling Parameters: These are already configured for the workflow in this tutorial. You do not need to configure them for inner nodes. They can be used directly in tasks or code.

   • Scheduling Policies: You can set the Time for Delayed Execution parameter to specify how long a child node waits to run after the workflow starts. You do not need to set this parameter for this tutorial.

5. In the top toolbar, click Save to save the current node.