Synchronize data

Create a MySQL data source ( user_behavior_analysis_mysql)

The user profile data for this tutorial is stored in a MySQL database. Create a MySQL data source to synchronize the user profile data (ods_user_info_d) to the OSS object storage that you created when you prepared the environment.

1. Go to the Data Sources page.

   1. Log on to the DataWorks console. In the target region, click More > Management Center in the left-side navigation pane. Select a workspace from the drop-down list and click Go to Management Center.

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

2. Click Add Connection, and then search for and select MySQL as the data source type.

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

   The following table describes the required parameters. For parameters not listed, keep the default values.

   Parameter	Description
   Data Source Name	Enter a name for the data source. For this tutorial, enter user_behavior_analysis_mysql.
   Description	This data source is for the DataWorks tutorial. It provides read-only access to sample data for single-table offline tasks within Data Integration and cannot be used by other modules.
   Configuration Mode	Select User-created Data Store with Public IP Addresses.
   Connection Address	Host IP address: rm-bp1z69dodhh85z9qa.mysql.rds.aliyuncs.com Port: 3306
   Database Name	Enter the database name. For this tutorial, enter workshop.
   Username	Enter the username. For this tutorial, enter workshop.
   Password	Enter the password. For this tutorial, enter workshop#2017.
   Authentication Method	No authentication.

4. In the Connection Configuration area, click Test Connectivity for the production and development environments, and ensure that the connectivity status is Connectable.

   Important

   • Make sure that the resource group is bound to the workspace and that public network access is enabled. Otherwise, data synchronization will fail. For the configuration steps, see Prepare the environment.

   • If you do not have an available resource group, you can follow the instructions on the page and click Go to Buy and Associate Purchased Resource Group.

5. Click Complete Creation.

Create an HttpFile data source ( user_behavior_analysis_httpfile)

The user website access logs for this tutorial are stored in OSS. Create an HttpFile data source to synchronize the user website access logs (user_log.txt) to the OSS object storage that you created when you prepared the environment.

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

2. Click Add Connection. In the Add Connection dialog box, search for and select HttpFile as the data source type.

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

   The following table describes the required parameters. For parameters not listed, keep the default values.

   Parameter	Description
   Data Source Name	Enter a name for the data source. For this tutorial, enter user_behavior_analysis_httpfile.
   Description	This data source is for the DataWorks tutorial. It provides read-only access to sample data for single-table offline tasks within Data Integration and cannot be used by other modules.
   URL	Set the URL for both the development and production environments to https://dataworks-workshop-2024.oss-cn-shanghai.aliyuncs.com.

4. In the Connection Configuration area, click Test Connectivity for the production and development environments, and ensure that the connectivity status is Connectable.

   Important

   • Make sure that the resource group is bound to the workspace and that public network access is enabled. Otherwise, data synchronization will fail. For the configuration steps, see Prepare the environment.

   • If you do not have an available resource group, you can follow the instructions on the page and click Go to Buy and Associate Purchased Resource Group.

5. Click Complete Creation.

Create an OSS data source

In this tutorial, you synchronize user profiles from the MySQL data source and logs from the HttpFile data source to an OSS data source.

1. On the Management Center page, click Data Source, and then click Add Connection.

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

3. On the Create OSS Data Source page, configure the parameters. In this tutorial, use the example values for both the development environment and production environment.

   Parameter	Description
   Data Source Name	Enter a name for the data source. In this example, the name is test_g.
   Description	Enter a brief description of 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 it is created and cannot be viewed again later. Keep it safe. If the AccessKey is leaked or lost, delete it and create a new AccessKey.
   Endpoint	Enter http://oss-cn-shanghai-internal.aliyuncs.com.
   Bucket	The name of the private OSS bucket that you prepared during environment preparation. In this example, the name is dw-spark-demo.

4. Click Test Connectivity in the Connection Status (Development Environment) and Connection Status (Production Environment) columns of the specified resource group, and wait until the connectivity test is complete and the status shows Connectable.

   Note

   Make sure that at least one resource group is in the Connectable state. Otherwise, you cannot use wizard mode to create synchronization tasks for this data source.

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

Configure the initial node

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

2. On the right side of the workshop_start_spark node editing page, click Scheduling Configuration and configure the parameters. The following table describes the key parameters required for this example. Retain the default values for parameters that are not described.

   Schedule settings parameter	Description
   Scheduling Resource Groups	In this tutorial, select the serverless resource group that you created during the environment preparation stage.
   Node dependency settings	Because workshop_start_spark is the initial node and has no upstream dependency, you can click Using the Workspace Root Node to trigger the workflow execution from the workspace root node. The workspace root node is named workspace name_root.

Configure the user log synchronization pipeline (ods_raw_log_d_2oss_spark)

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

1. On the workflow canvas, 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 resource settings.

   Parameter	Description
   Data source	Source: HttpFile. Data source name: user_behavior_analysis_httpfile.
   My Resource Group	Select the serverless resource group that you purchased during the environment preparation stage.
   Data going	Destination: OSS. Data source name: Select the private OSS data source that you created earlier. In this example, it is test_g.

   Important

   If the network is not connected, check whether public network access is enabled for the serverless resource group.

3. Click The next Step to configure the synchronization task.

   • Configure Source and Destination

     The following table describes the key parameters required for this example. Retain the default values for parameters that are not described.

     Parameter	Description
     Data source	File Path: /user_log.txt. File Type: Select text. Field Delimiter: Enter the column delimiter |. Compression Format: Select None. Skip Header: Select No.
     Data going	File Type: Select text. File Name (including path): Enter the path based on the directory structure of your OSS bucket. In this example, the value is ods_raw_log_d/log_${bizdate}/log_${bizdate}.txt, where ods_raw_log_d is the name of the directory that you created, and $bizdate represents the date of the previous day. Field Delimiter: Enter the column delimiter |.

   • Confirm the Field Mapping and Channel settings.

     DataWorks uses the configured field mapping to write data from specified source fields to corresponding destination fields. You can also configure settings such as task concurrency and a dirty data policy. For this tutorial, set the Policy for Dirty Data Records to Disallow Dirty Data Records, and leave other settings at their default values. For more information, see Configure a sync task in wizard mode.

4. Configure the debug parameters.

   On the right side of the single-table batch synchronization task configuration page, click Run Configuration and configure the following parameters. These parameters are used for debugging in Step 4 with the Run Configuration parameters for test runs.

   Configuration item	Description
   Resource Group	Select the serverless resource group that you purchased during the environment preparation stage.
   Script Parameters	Click Add parameter and set bizdate to a value in the yyyymmdd format (for example, bizdate=20250223). During debugging, Data Studio uses this constant to replace the variable defined in the task.

5. (Optional) Configure scheduling properties.

   In this tutorial, retain the default values for the schedule settings. On the right side of the node editing page, click Scheduling Configuration. For more information about the parameters, see Schedule settings.

   • Scheduling Parameters: In this tutorial, scheduling parameters are configured at the workflow level. You do not need to configure them for individual nodes within the workflow. The parameters can be used directly in tasks or code.

   • Scheduling Policy: You can specify the Delayed execution time parameter to define how long a child node waits before it starts running after the workflow starts. This parameter is not configured in this tutorial.

6. Click Save on the top toolbar to save the current node.

Configure the user data synchronization pipeline (ods_user_info_d_2oss_spark)

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

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

2. Configure the network and resource settings for the synchronization pipeline.

   Parameter	Description
   Data source	Source: MySQL. Data source name: user_behavior_analysis_mysql.
   My Resource Group	Select the serverless resource group that you purchased during the environment preparation stage.
   Data going	Destination: OSS. Data source name: Select the private OSS data source that you created earlier. In this example, it is test_g.

   Important

   If the network is not connected, check whether public network access is enabled for the serverless resource group.

3. Click The next Step to configure the synchronization task.

   • Configure Source and Destination

     The following table describes the key parameters required for this example. Retain the default values for parameters that are not described.

     Parameter	Description
     Data source	Table: Select ods_user_info_d from the data source. Shard Key: We recommend that you use a primary key or an indexed column as the split key. Only integer columns are supported. In this example, set the split key to uid.
     Data going	File Type: Select text. File Name (including path): Enter the path based on the directory structure of your OSS bucket. In this example, the value is ods_user_info_d/user_${bizdate}/user_${bizdate}.txt, where ods_user_info_d is the name of the directory that you created, and $bizdate represents the date of the previous day. Field Delimiter: Enter the column delimiter |.

   • Confirm the Field Mapping and Channel settings.

     DataWorks uses the configured field mapping to write data from specified source fields to corresponding destination fields. You can also configure settings such as task concurrency and a dirty data policy. For this tutorial, set the Policy for Dirty Data Records to Disallow Dirty Data Records, and leave other settings at their default values. For more information, see Configure a sync task in wizard mode.

4. Configure the debug parameters.

   On the right side of the single-table batch synchronization task configuration page, click Run Configuration and configure the following parameters. These parameters are used for debugging in Step 4 with the Run Configuration parameters for test runs.

   Configuration item	Description
   Resource Group	Select the serverless resource group that you purchased during the environment preparation stage.
   Script Parameters	Click Add parameter and set bizdate to a value in the yyyymmdd format (for example, bizdate=20250223). During debugging, Data Studio uses this constant to replace the variable defined in the task.

5. (Optional) Configure scheduling properties.

   In this tutorial, retain the default values for the schedule settings. On the right side of the node editing page, click Scheduling Configuration. For more information about the parameters, see Schedule settings.

   • Scheduling Parameters: In this tutorial, scheduling parameters are configured at the workflow level. You do not need to configure them for individual nodes within the workflow. The parameters can be used directly in tasks or code.

   • Scheduling Policy: You can specify the Delayed execution time parameter to define how long a child node waits before it starts running after the workflow starts. This parameter is not configured in this tutorial.

6. Click Save on the top toolbar to save the current node.

Create the log table (ods_raw_log_d_spark) and parse data

After data is synchronized to the private OSS data source through the batch synchronization task, the data is written to the external table ods_raw_log_d_spark created by the EMR SPARK SQL node based on the generated OSS files.

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

2. Write the table creation statement.

   Paimon table (DLF)

   -- 1. Create an Append-Only Paimon table
   CREATE TABLE IF NOT EXISTS ods_raw_log_d_spark (
     `col` STRING,
     `dt` STRING   -- Include the partition key as a regular column, which is the recommended approach for Paimon
   )
   PARTITIONED BY (dt)
   TBLPROPERTIES (
     'format' = 'paimon'
   );
   
   -- 2. Create a temporary view that points to and parses the source files on OSS
   CREATE TEMPORARY VIEW source_of_logs
   (
       -- The view has only one column to read the entire line of text
       `value` STRING 
   )
   USING TEXT
   OPTIONS (
       path 'oss://dw-spark-demo.oss-cn-shanghai-internal.aliyuncs.com/ods_raw_log_d/log_${bizdate}/'
   );
   
   INSERT INTO ods_raw_log_d_spark
   SELECT 
     value,      -- Your raw log line
     '${bizdate}' AS dt -- Specify the partition value directly in the SELECT statement
   FROM source_of_logs;

   Hive table (DLF-Legacy)

   -- Scenario: The following Spark SQL creates an external table ods_raw_log_d_spark via EMR Spark SQL, using LOCATION to read log data written by an offline data integration task to a private OSS bucket, and adds the corresponding dt partition.
   -- Notes:
   --      DataWorks provides scheduling parameters to write daily incremental data into the corresponding business partition of the target table during scheduling.
   --      In development, you can define code variables in the ${variable_name} format and assign scheduling parameters to them on the scheduling configuration page, enabling dynamic parameter injection 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 preceding code based on your actual environment. dw-spark-demo is the name of the OSS bucket that you created when you prepared the environment.

3. Configure the debug parameters.

   On the right side of the EMR SPARK SQL task configuration page, click Run Configuration and configure the following parameters. These parameters are used for debugging in Step 4 with the Run Configuration parameters for test runs.

   Configuration item	Description
   Computing Resources	Select the Spark compute resource that you associated during the environment preparation stage.
   Resource Group	Select the serverless resource group that you purchased during the environment preparation stage.
   Script Parameters	Click Add parameter and set bizdate to a value in the yyyymmdd format (for example, bizdate=20250223). During debugging, Data Studio uses this constant to replace the variable defined in the task.

4. (Optional) Configure scheduling properties.

   In this tutorial, retain the default values for the schedule settings. On the right side of the node editing page, click Scheduling Configuration. For more information about the parameters, see Schedule settings.

   • Scheduling Parameters: In this tutorial, scheduling parameters are configured at the workflow level. You do not need to configure them for individual nodes within the workflow. The parameters can be used directly in tasks or code.

   • Scheduling Policy: You can specify the Delayed execution time parameter to define how long a child node waits before it starts running after the workflow starts. This parameter is not configured in this tutorial.

5. Click Save on the top toolbar to save the current node.

Create the user table (ods_user_info_d_spark) and parse data

After data is synchronized to the private OSS data source through the batch synchronization task, the data is written to the external table ods_user_info_d_spark created by the EMR SPARK SQL node based on the generated OSS files.

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

2. Configure the network and resource settings for the synchronization pipeline.

   Paimon table (DLF)

   -- 1. Create a Paimon table as the target
   CREATE 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',
       `dt`         STRING COMMENT 'Partition date'
   ) 
   PARTITIONED BY (dt)
   TBLPROPERTIES (
       'format' = 'paimon'
   );
   
   -- 2. Create a temporary view that points to and parses the source files on OSS
   CREATE TEMPORARY VIEW source_of_user_info
   (
       -- The view has only one column to read the entire line of text
       `value` STRING 
   )
   USING TEXT -- Tell Spark this is a plain text file
   OPTIONS (
       path 'oss://dw-spark-demo.oss-cn-shanghai-internal.aliyuncs.com/ods_user_info_d/user_${bizdate}/'
   );
   
   -- 3. Query and parse data from the temporary view, then insert into the Paimon table
   INSERT INTO ods_user_info_d_spark
   SELECT 
       -- Use the split function to split the raw text line by '|'
       split(value, '\\|')[0] AS uid,
       split(value, '\\|')[1] AS gender,
       split(value, '\\|')[2] AS age_range,
       split(value, '\\|')[3] AS zodiac,
       '${bizdate}' AS dt -- Assign the value for the partition field
   FROM 
       source_of_user_info;

   Hive table (DLF-Legacy)

   -- Scenario: The following Spark SQL creates an external table ods_user_info_d_spark via EMR Spark SQL, using LOCATION to read user information written by an offline data integration task to a private OSS bucket, and writes data into the corresponding dt partition.
   -- Notes:
   --      DataWorks provides scheduling parameters to write daily incremental data into the corresponding business partition of the target table during scheduling.
   --      In development, you can define code variables in the ${variable_name} format and assign scheduling parameters to them on the scheduling configuration page, enabling dynamic parameter injection 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 preceding code based on your actual environment. dw-spark-demo is the name of the OSS bucket that you created when you prepared the environment.

3. Configure the debug parameters.

   On the right side of the EMR SPARK SQL task configuration page, click Run Configuration and configure the following parameters. These parameters are used for debugging in Step 4 with the Run Configuration parameters for test runs.

   Configuration item	Description
   Computing Resources	Select the Spark compute resource that you associated during the environment preparation stage.
   Resource Group	Select the serverless resource group that you purchased during the environment preparation stage.
   Script Parameters	Click Add parameter and set bizdate to a value in the yyyymmdd format (for example, bizdate=20250223). During debugging, Data Studio uses this constant to replace the variable defined in the task.

4. (Optional) Configure scheduling properties.

   In this tutorial, retain the default values for the schedule settings. On the right side of the node editing page, click Scheduling Configuration. For more information about the parameters, see Schedule settings.

   • Scheduling Parameters: In this tutorial, scheduling parameters are configured at the workflow level. You do not need to configure them for individual nodes within the workflow. The parameters can be used directly in tasks or code.

   • Scheduling Policy: You can specify the Delayed execution time parameter to define how long a child node waits before it starts running after the workflow starts. This parameter is not configured in this tutorial.

5. Click Save on the top toolbar to save the current node.