DataWorks:Process data

Upload a resource (ip2region-starrocks.jar)

1. Download the ip2region-starrocks.jar package.

   Note

   The ip2region-starrocks.jar package is used only in this tutorial.

2. Upload a resource to OSS.

   1. Log on to the OSS console and go to the Buckets page. On the Buckets page, find the bucket that is created when you prepare environments and create the dataworks_starrocks directory in the path of the bucket.

   2. Upload the ip2region-starrocks.jar package to the dataworks_starrocks directory.

      In this tutorial, the full path in which the package is stored is https://test.oss-cn-shanghai-internal.aliyuncs.com/dataworks_starrocks/ip2region-starrocks.jar. You can obtain the path in which the uploaded OSS resource is stored based on the full path.

      Note

      • In this tutorial, a bucket named test is used.

      • The network address of the bucket to which the UDF belongs is the address that is used for Access from ECS over the Classic Network (internal network).

      • If you use an internal endpoint, the OSS bucket must reside in the same region as the DataWorks workspace. In this example, the China (Shanghai) region is used.

Register a function (getregion)

1. Create a StarRocks node to register a function.

   Log on to the DataWorks console and go to the DATA STUDIO pane of the Data Studio page. In the Workspace Directories section of the DATA STUDIO pane, click the icon and choose Create Node > Database > StarRocks to create a StarRocks node.

2. Write code to register a function.

   • Register a function.

     CREATE FUNCTION getregion(string)
     RETURNS string
     PROPERTIES ( 
         "symbol" = "com.starrocks.udf.sample.Ip2Region", 
         "type" = "StarrocksJar",
         "file" = "Enter the full storage path of the OSS bucket. You can obtain the path in the preceding substep."
     );

   • Check whether the function is registered.

     SELECT getregion('The IP address of your on-premises machine');

   Important

   A function can be separately registered in the development environment and production environment only once. You must deploy the StarRocks node to the production environment before you can register a function in the production environment.

3. In the top toolbar of the configuration tab of the StarRocks node, click Save. Then, click Deploy to deploy the StarRocks node to the StarRocks computing resource in the development environment and production environment by following the instructions displayed on the DEPLOY tab. Then, backfill data for the StarRocks node to complete the registration of the function in the production environment. After the function is registered, manually freeze the StarRocks node in the production environment in Operation Center. This prevents the StarRocks node from failing due to repeated registration.

Configure the dwd_log_info_di_starrocks node

In the sample code for this node, the registered function is used to process the SQL code for fields in the ancestor table ods_raw_log_d_starrocks and synchronize the data in the table to the dwd_log_info_di_starrocks table.

1. In the canvas of the workflow, move the pointer over the dwd_log_info_di_starrocks node and click Open Node.

2. On the configuration tab of the node, select the StarRocks computing resource that is associated with the workspace when you prepare environments from the Select DataSource drop-down list.

3. Copy the following SQL statements and paste them in the code editor:

   Note

   In the sample code for the dwd_log_info_di_starrocks node, the registered function is used to process the SQL code for fields in the ancestor table ods_raw_log_d_starrocks and synchronize the data in the table to the dwd_log_info_di_starrocks table.

   Sample code for the dwd_log_info_di_starrocks node

   CREATE TABLE IF NOT EXISTS dwd_log_info_di_starrocks (
       uid STRING COMMENT 'The user ID',
       ip STRING COMMENT 'The IP address',
       TIME STRING COMMENT 'The time in the yyyymmddhh:mi:ss format',
       status STRING COMMENT 'The status code that is returned by the server',
       bytes STRING COMMENT 'The number of bytes that are returned to the client',
       region STRING COMMENT 'The region, which is obtained based on the IP address',
       method STRING COMMENT 'The HTTP request type',
       url STRING COMMENT 'url',
       protocol STRING COMMENT 'The version number of HTTP',
       referer STRING COMMENT 'The source URL',
       device STRING COMMENT 'The terminal type',
       identity STRING COMMENT 'The access type, which can be crawler, feed, user, or unknown',
       dt DATE NOT NULL COMMENT 'The time'
   ) DUPLICATE KEY(uid) 
   COMMENT 'User behavior analysis case - The fact table for the user website access logs' 
   PARTITION BY(dt) 
   PROPERTIES ("replication_num" = "1");
   
   -- In this example, field data is dynamically partitioned based on the dt field. To prevent the node from rerunning and repeatedly writing data, the following SQL statement is used to delete the existing destination partitions before each processing operation. 
   ALTER TABLE dwd_log_info_di_starrocks DROP PARTITION IF EXISTS p${var} FORCE;
   
   -- Scenario: The following SQL statements use the getregion function to parse the IP address in the raw log data, and use regular expressions to split the raw data to analyzable fields and write the fields to the dwd_log_info_di_starrocks table. 
   -- Note:
   --     1. Before you can use a UDF on a DataWorks node, you must register a function. 
   --     2. You can configure scheduling parameters for nodes in DataWorks to synchronize incremental data to the related partition in the desired table every day in the scheduling scenario. 
   --        In actual development scenarios, you can define variables in the code of a node in the ${Variable name} format and assign scheduling parameters to the variables on the Properties tab of the configuration tab of the node. This way, the values of scheduling parameters can be dynamically replaced in the node code based on the configurations of the scheduling parameters. 
   INSERT INTO dwd_log_info_di_starrocks 
   SELECT 
       uid
       , ip  
       , time
       , status
       , bytes 
       , getregion(ip) AS region -- Obtain the region based on the IP address by using the UDF.
       ,REGEXP_EXTRACT(request, '([^ ]+)', 1) AS method
       ,REGEXP_EXTRACT(request, '^[^ ]+ (.*) [^ ]+$', 1) AS url
       ,REGEXP_EXTRACT(request, '([^ ]+)$', 1) AS protocol
       ,REGEXP_EXTRACT(referer, '^[^/]+://([^/]+)', 1) AS referer
     , CASE
       WHEN LOWER(agent) REGEXP 'android' THEN 'android'
       WHEN LOWER(agent) REGEXP 'iphone' THEN 'iphone'
       WHEN LOWER(agent) REGEXP 'ipad' THEN 'ipad'
       WHEN LOWER(agent) REGEXP 'macintosh' THEN 'macintosh'
       WHEN LOWER(agent) REGEXP 'windows phone' THEN 'windows_phone'
       WHEN LOWER(agent) REGEXP 'windows' THEN 'windows_pc'
       ELSE 'unknown'
   END AS device
     , CASE
       WHEN LOWER(agent) REGEXP '(bot|spider|crawler|slurp)' THEN 'crawler'
       WHEN LOWER(agent) REGEXP 'feed' OR REGEXP_EXTRACT(request, '^[^ ]+ (.*) [^ ]+$', 0) REGEXP 'feed' THEN 'feed'
       WHEN NOT (LOWER(agent) REGEXP '(bot|spider|crawler|feed|slurp)') 
            AND agent REGEXP '^(Mozilla|Opera)' 
            AND NOT (REGEXP_EXTRACT(request, '^[^ ]+ (.*) [^ ]+$', 0) REGEXP 'feed') THEN 'user'
       ELSE 'unknown'
   END AS identity,
    cast('${var}' AS DATE )AS dt
     FROM (
       SELECT
         SPLIT_PART(CAST(col AS VARCHAR(65533)), '##@@', 1)  AS ip
       , SPLIT_PART(CAST(col AS VARCHAR(65533)), '##@@', 2)  AS uid
       , SPLIT_PART(CAST(col AS VARCHAR(65533)), '##@@', 3)  AS time
       , SPLIT_PART(CAST(col AS VARCHAR(65533)), '##@@', 4)  AS request
       , SPLIT_PART(CAST(col AS VARCHAR(65533)), '##@@', 5)  AS status
       , SPLIT_PART(CAST(col AS VARCHAR(65533)), '##@@', 6)  AS bytes
       , SPLIT_PART(CAST(col AS VARCHAR(65533)), '##@@', 7)  AS referer
       , SPLIT_PART(CAST(col AS VARCHAR(65533)), '##@@', 8)  AS agent
   FROM
       ods_raw_log_d_starrocks
   WHERE
       dt = '${var}'
   ) a;

4. Configure debugging parameters.

   In the right-side navigation pane of the configuration tab of the node, click Debugging Configurations. On the Debugging Configurations tab, configure the following parameters. These parameters are used to test the workflow in Step 4.

   Parameter	Description
   Computing Resource	Select the StarRocks computing resource that is associated with the workspace when you prepare environments.
   Resource Group	Select the serverless resource group that you purchase when you prepare environments.
   Script Parameters	In the Parameter Value column of the var parameter, enter a constant value in the yyyymmdd format. Example: var=20250223. When you debug the workflow, Data Studio replaces the variables defined for nodes in the workflow with the constant.

5. Optional. Configure scheduling properties.

   You can retain default values for parameters related to scheduling properties in this tutorial. You can click Properties in the right-side navigation pane of the node configuration tab. For information about parameters on the Properties tab, see Scheduling properties.

   • Scheduling Parameters: In this tutorial, scheduling parameters are configured for the workflow. You do not need to configure scheduling parameters for the inner nodes of the workflow. The configured scheduling parameters can be directly used for code and tasks developed based on the inner nodes.

   • Scheduling Policies: You can configure the Time for Delayed Execution parameter to specify the duration by which the running of the batch synchronization node lags behind the running of the workflow. In this tutorial, you do not need to configure this parameter.

6. In the top toolbar of the configuration tab, click Save to save the node.

Configure the dws_user_info_all_di_starrocks node

This node is used to aggregate data in the basic user information table ods_user_info_d_starrocks and the log data table dwd_log_info_di_starrocks and synchronize the aggregation result to the dws_user_info_all_di_starrocks table.

1. In the canvas of the workflow, move the pointer over the dws_user_info_all_di_starrocks node and click Open Node.

2. On the configuration tab of the node, select the StarRocks computing resource that is associated with the workspace when you prepare environments from the Select DataSource drop-down list.

3. Copy the following SQL statements and paste them in the code editor:

   Note

   On the configuration tab of the dws_user_info_all_di_starrocks node, write code for aggregating data in the ancestor tables dwd_log_info_di_starrocks and ods_user_info_d_starrocks and synchronizing the aggregation result to the dws_user_info_all_di_starrocks table.

   Sample code for the dws_user_info_all_di_starrocks node

   CREATE TABLE IF NOT EXISTS dws_user_info_all_di_starrocks (
       uid STRING COMMENT 'The user ID',
       gender STRING COMMENT 'The gender',
       age_range STRING COMMENT 'The age range',
       zodiac STRING COMMENT 'The zodiac sign',
       region STRING COMMENT 'The region, which is obtained based on the IP address',
       device STRING COMMENT 'The terminal type',
       identity STRING COMMENT 'The access type, which can be crawler, feed, user, or unknown',
       method STRING COMMENT 'The HTTP request type',
       url STRING COMMENT 'url',
       referer STRING COMMENT 'The source URL',
       TIME STRING COMMENT 'The time in the yyyymmddhh:mi:ss format',
       dt DATE NOT NULL COMMENT 'The time'
   ) DUPLICATE KEY(uid) 
   COMMENT 'User behavior analysis case - Wide table for the website access information about users' 
   PARTITION BY(dt) 
   PROPERTIES ("replication_num" = "1");
   
   -- In this example, field data is dynamically partitioned based on the dt field. To prevent the node from rerunning and repeatedly writing data, the following SQL statement is used to delete the existing destination partitions before each processing operation. 
   ALTER TABLE dws_user_info_all_di_starrocks DROP PARTITION IF EXISTS p${var} FORCE;
   
   
   -- Scenario: Merge the processed log data stored in the dwd_log_info_di_starrocks table and the basic user information stored in the ods_user_info_d_starrocks table and synchronize the merged data to the dws_user_info_all_di_starrocks table. 
   -- Note: You can configure scheduling parameters for nodes in DataWorks to synchronize incremental data to the related partition in the desired table every day in the scheduling scenario. 
   -- In actual development scenarios, you can define variables in the code of a node in the ${Variable name} format and assign scheduling parameters to the variables on the Properties tab of the configuration tab of the node. This way, the values of scheduling parameters can be dynamically replaced in the node code based on the configurations of the scheduling parameters. 
   INSERT INTO dws_user_info_all_di_starrocks 
   SELECT 
       IFNULL(a.uid, b.uid) AS uid,
       b.gender,
       b.age_range,
       b.zodiac,
       a.region,
       a.device,
       a.identity,
       a.method,
       a.url,
       a.referer,
       a.time,
       a.dt
   FROM dwd_log_info_di_starrocks a
   LEFT JOIN ods_user_info_d_starrocks b
   ON a.uid = b.uid
   WHERE a.dt = '${var}';
   

4. Configure debugging parameters.

   In the right-side navigation pane of the configuration tab of the node, click Debugging Configurations. On the Debugging Configurations tab, configure the following parameters. These parameters are used to test the workflow in Step 4.

   Parameter	Description
   Computing Resource	Select the StarRocks computing resource that is associated with the workspace when you prepare environments.
   Resource Group	Select the serverless resource group that you purchase when you prepare environments.
   Script Parameters	In the Parameter Value column of the var parameter, enter a constant value in the yyyymmdd format. Example: var=20250223. When you debug the workflow, Data Studio replaces the variables defined for nodes in the workflow with the constant.

5. Optional. Configure scheduling properties.

   You can retain default values for parameters related to scheduling properties in this tutorial. You can click Properties in the right-side navigation pane of the node configuration tab. For information about parameters on the Properties tab, see Scheduling properties.

   • Scheduling Parameters: In this tutorial, scheduling parameters are configured for the workflow. You do not need to configure scheduling parameters for the inner nodes of the workflow. The configured scheduling parameters can be directly used for code and tasks developed based on the inner nodes.

   • Scheduling Policies: You can configure the Time for Delayed Execution parameter to specify the duration by which the running of the batch synchronization node lags behind the running of the workflow. In this tutorial, you do not need to configure this parameter.

6. In the top toolbar of the configuration tab, click Save to save the node.

Configure the ads_user_info_1d_starrocks node

This node is used to further process data in the dws_user_info_all_di_starrocks table and synchronize the processed data to the ads_user_info_1d_starrocks table to generate a basic user profile.

1. In the canvas of the workflow, move the pointer over the ads_user_info_1d_starrocks node and click Open Node.

2. On the configuration tab of the node, select the StarRocks computing resource that is associated with the workspace when you prepare environments from the Select DataSource drop-down list.

3. Copy the following SQL statements and paste them in the code editor:

   Sample code for the ads_user_info_1d_starrocks node

   CREATE TABLE IF NOT EXISTS ads_user_info_1d_starrocks (
   uid STRING COMMENT 'The user ID',
   region STRING COMMENT 'The region, which is obtained based on the IP address',
   device STRING COMMENT 'The terminal type',
   pv BIGINT COMMENT 'pv',
   gender STRING COMMENT 'The gender',
   age_range STRING COMMENT 'The age range',
   zodiac STRING COMMENT 'The zodiac sign',
   dt DATE NOT NULL COMMENT 'The time'
   ) DUPLICATE KEY(uid) 
   COMMENT 'User behavior analysis case - User profile data' 
   PARTITION BY(dt) 
   PROPERTIES ("replication_num" = "1");
   
   -- In this example, field data is dynamically partitioned based on the dt field. To prevent the node from rerunning and repeatedly writing data, the following SQL statement is used to delete the existing destination partitions before each processing operation. 
   ALTER TABLE ads_user_info_1d_starrocks DROP PARTITION IF EXISTS p${var} FORCE;
   
   -- Scenario: The following SQL statements are used to further process data in the dws_user_info_d_all_di_starrocks wide table that stores the user website access logs to generate basic user profile data, and synchronize the data to the ads_user_info_1d_starrocks table. 
   -- Note: You can configure scheduling parameters for nodes in DataWorks to synchronize incremental data to the related partition in the desired table every day in the scheduling scenario. 
   -- In actual development scenarios, you can define variables in the code of a node in the ${Variable name} format and assign scheduling parameters to the variables on the Properties tab of the configuration tab of the node. This way, the values of scheduling parameters can be dynamically replaced in the node code based on the configurations of the scheduling parameters. 
   INSERT INTO ads_user_info_1d_starrocks 
   SELECT 
   uid,
   MAX(region) AS region,
   MAX(device) AS device,
   COUNT(*) AS pv,
   MAX(gender) AS gender,
   MAX(age_range) AS age_range,
   MAX(zodiac) AS zodiac,
   dt
   FROM dws_user_info_all_di_starrocks
   WHERE dt = '${var}'
   GROUP BY uid, dt;
   
   SELECT * FROM dws_user_info_all_di_starrocks
   WHERE dt = '${var}';

4. Configure debugging parameters.

   In the right-side navigation pane of the configuration tab of the node, click Debugging Configurations. On the Debugging Configurations tab, configure the following parameters. These parameters are used to test the workflow in Step 4.

   Parameter	Description
   Computing Resource	Select the StarRocks computing resource that is associated with the workspace when you prepare environments.
   Resource Group	Select the serverless resource group that you purchase when you prepare environments.
   Script Parameters	In the Parameter Value column of the var parameter, enter a constant value in the yyyymmdd format. Example: var=20250223. When you debug the workflow, Data Studio replaces the variables defined for nodes in the workflow with the constant.

5. Optional. Configure scheduling properties.

   You can retain default values for parameters related to scheduling properties in this tutorial. You can click Properties in the right-side navigation pane of the node configuration tab. For information about parameters on the Properties tab, see Scheduling properties.

   • Scheduling Parameters: In this tutorial, scheduling parameters are configured for the workflow. You do not need to configure scheduling parameters for the inner nodes of the workflow. The configured scheduling parameters can be directly used for code and tasks developed based on the inner nodes.

   • Scheduling Policies: You can configure the Time for Delayed Execution parameter to specify the duration by which the running of the batch synchronization node lags behind the running of the workflow. In this tutorial, you do not need to configure this parameter.

6. In the top toolbar of the configuration tab, click Save to save the node.