E-MapReduce:Synchronize data

Add a MySQL data source

Add the MySQL database as a data source so you can synchronize the ods_user_info_d table to the OSS bucket.

1. Go to the Data Sources page.

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

   2. In the left-side navigation pane of the SettingCenter page, click Data Sources.

2. On the Data Sources page, click Add Data Source. In the Add Data Source dialog box, search for and select MySQL.

3. On the Add MySQL Data Source page, configure the parameters. Set Data Source Name to user_behavior_analysis_mysql.

4. In the Connection Configuration section, find the resource group and click Test Network Connectivity in both the Connection Status (Development Environment) and Connection Status (Production Environment) columns. Make sure the status shows Connected.

   Important

   The resource group must be associated with your workspace and have internet access. If no resource group is available, click Purchase or Associate Purchased Resource Group in the prompt message to resolve this.

5. Click Complete Creation.

Add an HttpFile data source

Add an HttpFile data source to synchronize the user_log.txt access logs from OSS into your private OSS bucket.

1. In the left-side navigation pane of the SettingCenter page, click Data Sources. On the Data Sources page, click Add Data Source.

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

3. On the Add HttpFile Data Source page, configure the following parameters. Retain default values for other parameters.

   Parameter	Value
   Data Source Name	user_behavior_analysis_httpfile
   Data Source Description	The data source is exclusive for experiencing cases in DataWorks. You can configure the data source in a batch synchronization task to access the test data provided by the system. The data source can be used only for data reading in Data Integration and cannot be used in other services.
   URL	https://dataworks-workshop-2024.oss-cn-shanghai.aliyuncs.com (set for both development and production environments)

4. In the Connection Configuration section, click Test Network Connectivity in both the Connection Status (Development Environment) and Connection Status (Production Environment) columns for the target resource group. Confirm that the status shows Connected.

   Important

   The resource group must be associated with your workspace and have internet access. If no resource group is available, click Purchase or Associate Purchased Resource Group in the prompt message.

5. Click Complete Creation.

Add an OSS data source

Add a private OSS data source as the destination for synchronized data.

1. In the left-side navigation pane of the SettingCenter page, click Data Sources. In the upper-left corner, click Add Data Source.

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

3. On the Add OSS Data Source page, configure the following parameters. Use the same values for both development and production environments.

   Parameter	Value
   Data Source Name	test_g
   Access Mode	Select AccessKey Mode
   AccessKey ID	The AccessKey ID of your account. Go to the AccessKey page to copy it.
   AccessKey Secret	The AccessKey secret of your account. The secret is only displayed at creation time—store it securely. If the AccessKey pair is leaked, delete it and create a new one.
   Endpoint	http://oss-cn-shanghai-internal.aliyuncs.com
   Bucket	dw-spark-demo (the private OSS bucket created during environment preparation)

4. Click Test Network Connectivity in both the Connection Status (Development Environment) and Connection Status (Production Environment) columns. Make sure at least one resource group shows Connected—otherwise you won't be able to configure synchronization nodes with the codeless UI.

5. Click Complete Creation.

Configure the workshop_start_spark node

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

2. In the right-side navigation pane, click Properties and configure the following parameters.

   Parameter	Value
   Scheduling Type	Dry-run
   Resource Group For Scheduling	The serverless resource group created during environment preparation.
   Scheduling Dependencies	Click Use Workspace Root Node to trigger from the workspace root node (format: Workspace name_root).

Configure the ods_raw_log_d_2oss_spark node

This node synchronizes website access logs from the HttpFile data source to the private OSS bucket.

1. In the canvas, hover over the ods_raw_log_d_2oss_spark node and click Open Node.

2. Configure the source, destination, and resource group.

   Section	Setting
   Source	Source: HttpFile; Data Source Name: user_behavior_analysis_httpfile
   Resource Group	Select the serverless resource group created during environment preparation.
   Destination	Destination: OSS; Data Source Name: test_g

3. Click Next and configure the source and destination settings. In the Field Mapping section, confirm field mappings. In the Channel Control section, set Policy for Dirty Data Records to Disallow Dirty Data Records. Retain defaults for other parameters. For details, see Configure a batch synchronization task by using the codeless UI.

   Section	Parameter	Value
   Source	File Path	/user_log.txt
   	Text type	text
   	Column Delimiter	|
   	Compression format	None
   	Skip Header	No
   Destination	Text type	text
   	Object Name (Path Included)	ods_raw_log_d/log_${bizdate}/log_${bizdate}.txt
   	Column Delimiter	|

4. Configure debugging parameters. In the right-side navigation pane, click Debugging Configurations and set the following.

   Parameter	Value
   Resource Group	Select the serverless resource group.
   Script Parameters	Set bizdate to a date in yyyymmdd format. Example: bizdate=20250223.

5. (Optional) Click Properties to configure scheduling properties. Scheduling parameters are inherited from the workflow—no further configuration needed for inner nodes. If you want to offset the node's start time relative to the workflow, set Time for Delayed Execution under Scheduling Policies. For other parameters, see Node scheduling.

6. Click Save.

Configure the ods_user_info_d_2oss_spark node

This node synchronizes user profile data from the MySQL data source to the private OSS bucket.

1. In the canvas, hover over the ods_user_info_d_2oss_spark node and click Open Node.

2. Configure the source, destination, and resource group.

   Section	Setting
   Source	Source: MySQL; Data Source Name: user_behavior_analysis_mysql
   Resource Group	Select the serverless resource group associated with your workspace.
   Destination	Destination: OSS; Data Source Name: test_g

3. Click Next and configure the source and destination settings. In the Field Mapping section, confirm field mappings. In the Channel Control section, set Policy for Dirty Data Records to Disallow Dirty Data Records.

   Section	Parameter	Value
   Source	Table	ods_user_info_d
   	Split key	uid (use the primary key or an indexed column of INTEGER type as the shard key)
   Destination	Text Type	text
   	Object Name (Path Included)	ods_user_info_d/user_${bizdate}/user_${bizdate}.txt
   	Column Delimiter	|

4. Configure debugging parameters. In the right-side navigation pane, click Debugging Configurations and set the following.

   Parameter	Value
   Resource Group	Select the serverless resource group.
   Script Parameters	Set bizdate to a date in yyyymmdd format. Example: bizdate=20250223.

5. (Optional) Configure scheduling properties as described in the ods_raw_log_d_2oss_spark section.

6. Click Save.

Configure the ods_raw_log_d_spark node

This EMR Spark SQL node creates the ods_raw_log_d_spark external table and points it to the access logs synchronized to the OSS bucket.

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

2. Enter the following SQL statements.

   Replace dw-spark-demo in the LOCATION value with your OSS bucket name from environment preparation.

   -- Creates the ods_raw_log_d_spark external table using EMR Spark SQL.
   -- LOCATION points to the access logs synchronized to the private OSS bucket.
   -- The $bizdate variable is replaced at runtime by the scheduling parameter configured for the workflow.
   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}/'

3. Configure debugging parameters. In the right-side navigation pane, click Debugging Configurations and set the following.

   Parameter	Value
   Computing Resource	Select the EMR Serverless Spark computing resource associated with your workspace during environment preparation.
   Resource Group	Select the serverless resource group.
   Script Parameters	Click Add Parameter. Enter bizdate as the parameter name and a date in yyyymmdd format as the value. Example: bizdate=20250223.

4. (Optional) Configure scheduling properties as described in the ods_raw_log_d_2oss_spark section.

5. Click Save.

Configure the ods_user_info_d_spark node

This EMR Spark SQL node creates the ods_user_info_d_spark external table and points it to the user profile data synchronized to the OSS bucket.

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

2. Enter the following SQL statements.

   Replace dw-spark-demo in the LOCATION value with your OSS bucket name from environment preparation.

   -- Creates the ods_user_info_d_spark external table using EMR Spark SQL.
   -- LOCATION points to the user profile data synchronized to the private OSS bucket.
   -- The $bizdate variable is replaced at runtime by the scheduling parameter configured for the workflow.
   CREATE EXTERNAL TABLE IF NOT EXISTS ods_user_info_d_spark
   (
       `uid`        STRING COMMENT 'The user ID'
       ,`gender`    STRING COMMENT 'The gender'
       ,`age_range` STRING COMMENT 'The age range'
       ,`zodiac`    STRING COMMENT 'The 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}/'
   ;

3. Configure debugging parameters. In the right-side navigation pane, click Debugging Configurations and set the following.

   Parameter	Value
   Computing Resource	Select the EMR Serverless Spark computing resource associated with your workspace.
   Resource Group	Select the serverless resource group.
   Script Parameters	Click Add Parameter. Enter bizdate as the parameter name and a date in yyyymmdd format as the value. Example: bizdate=20250223.

4. (Optional) Configure scheduling properties as described in the ods_raw_log_d_2oss_spark section.

5. Click Save.