DataWorks:Process data

Configure code

Double-click the dwd_log_info_di_spark node to open the configuration page and enter the following SQL statements.

-- Scenario: This Spark SQL script uses Spark SQL functions to split the ods_raw_log_d_spark table by "##@@" into multiple fields and writes the results to a new table named dwd_log_info_di_spark.
-- Note:
--      DataWorks provides scheduling parameters to write daily incremental data to the corresponding business partition of the target table.
--      In a real-world development scenario, you can define code variables by using the ${variable_name} format and assign scheduling parameters to them on the scheduling configuration page to enable dynamic parameter passing.  
CREATE TABLE IF NOT EXISTS dwd_log_info_di_spark (
  ip STRING COMMENT 'IP address',
  uid STRING COMMENT 'User ID',
  tm STRING COMMENT 'Time in yyyymmddhh:mi:ss format',
  status STRING COMMENT 'Status code returned by the server',
  bytes STRING COMMENT 'Bytes returned to the client',
  method STRING COMMENT'Request method',
  url STRING COMMENT 'URL',
  protocol STRING COMMENT 'Protocol',
  referer STRING ,
  device STRING,
  identity STRING
  )
PARTITIONED BY (
  dt STRING
);
ALTER TABLE dwd_log_info_di_spark ADD IF NOT EXISTS PARTITION (dt = '${bizdate}');
INSERT OVERWRITE TABLE dwd_log_info_di_spark PARTITION (dt='${bizdate}')
SELECT ip, 
       uid, 
       tm, 
       status, 
       bytes, 
       regexp_extract(request, '(^[^ ]+) .*', 1) AS method,
       regexp_extract(request, '^[^ ]+ (.*) [^ ]+$', 1) AS url,
       regexp_extract(request, '.* ([^ ]+$)', 1) AS protocol,
       regexp_extract(referer, '^[^/]+://([^/]+){1}', 1) AS referer,
       CASE 
           WHEN lower(agent) RLIKE 'android' THEN 'android' 
           WHEN lower(agent) RLIKE 'iphone' THEN 'iphone' 
           WHEN lower(agent) RLIKE 'ipad' THEN 'ipad' 
           WHEN lower(agent) RLIKE 'macintosh' THEN 'macintosh' 
           WHEN lower(agent) RLIKE 'windows phone' THEN 'windows_phone' 
           WHEN lower(agent) RLIKE 'windows' THEN 'windows_pc' 
           ELSE 'unknown' 
       END AS device, 
       CASE 
           WHEN lower(agent) RLIKE '(bot|spider|crawler|slurp)' THEN 'crawler' 
           WHEN lower(agent) RLIKE 'feed' OR regexp_extract(request, '^[^ ]+ (.*) [^ ]+$', 1) RLIKE 'feed' THEN 'feed' 
           WHEN lower(agent) NOT RLIKE '(bot|spider|crawler|feed|slurp)' AND agent RLIKE '^(Mozilla|Opera)' AND regexp_extract(request, '^[^ ]+ (.*) [^ ]+$', 1) NOT RLIKE 'feed' THEN 'user' 
           ELSE 'unknown' 
       END AS identity
FROM (
    SELECT 
        SPLIT(col, '##@@')[0] AS ip, 
        SPLIT(col, '##@@')[1] AS uid, 
        SPLIT(col, '##@@')[2] AS tm, 
        SPLIT(col, '##@@')[3] AS request, 
        SPLIT(col, '##@@')[4] AS status, 
        SPLIT(col, '##@@')[5] AS bytes, 
        SPLIT(col, '##@@')[6] AS referer, 
        SPLIT(col, '##@@')[7] AS agent
    FROM ods_raw_log_d_spark
    WHERE dt = '${bizdate}'
) a;

Configure scheduling properties

Save the configuration

Configure other required parameters as needed. When you are finished, click the icon in the toolbar of the node configuration tab to save the settings.

Verify the log table splitting

After the upstream nodes and the current node have run successfully, create an EMR Spark SQL Ad Hoc Query task to verify that the table was created correctly.

-- You must update the partition filter to your actual business date. For example, if the task runs on 2023-02-22, the business date is 2023-02-21 (the day before the task run date).
SELECT * FROM dwd_log_info_di_spark WHERE dt = 'your_business_date';

Edit code

Double-click the dws_user_info_all_di_spark node to open the node configuration page. Enter the following SQL statements.

-- Scenario: This Spark SQL script joins the dwd_log_info_di_spark and ods_user_info_d_spark tables on the uid field and writes the result to the corresponding dt partition.
-- Note:
--      DataWorks provides scheduling parameters to write daily incremental data to the corresponding business partition of the target table.
--      In a real-world development scenario, you can define code variables by using the ${variable_name} format and assign scheduling parameters to them on the scheduling configuration page to enable dynamic parameter passing.
CREATE TABLE IF NOT EXISTS dws_user_info_all_di_spark (
    uid        STRING COMMENT 'User ID',
    gender     STRING COMMENT 'Gender',
    age_range  STRING COMMENT 'Age range',
    zodiac     STRING COMMENT 'Zodiac sign',
    device     STRING COMMENT 'Terminal type',
    method     STRING COMMENT 'HTTP request type',
    url        STRING COMMENT 'URL',
    `time`     STRING COMMENT 'Time in yyyymmddhh:mi:ss format'
)
PARTITIONED BY (dt STRING);
-- Add a partition.
ALTER TABLE dws_user_info_all_di_spark ADD IF NOT EXISTS PARTITION (dt = '${bizdate}');
-- Insert data from the user and log tables.
INSERT OVERWRITE TABLE dws_user_info_all_di_spark PARTITION (dt = '${bizdate}')
SELECT 
    COALESCE(a.uid, b.uid) AS uid,
    b.gender AS gender,    
    b.age_range AS age_range,
    b.zodiac AS zodiac,
    a.device AS device,
    a.method AS method,
    a.url AS url,
    a.tm
FROM (
  SELECT * 
  FROM dwd_log_info_di_spark 
  WHERE dt='${bizdate}'
) a
LEFT OUTER JOIN (
  SELECT * 
  FROM ods_user_info_d_spark 
  WHERE dt='${bizdate}'
) b
ON 
    a.uid = b.uid;

Configure scheduling properties

Save the configuration

Configure other required parameters as needed. When you are finished, click the icon in the toolbar of the node configuration tab to save the settings.

Verify the detailed user log table data

After the upstream nodes and the current node have run successfully, create an EMR Spark SQL Ad Hoc Query task to verify that the table was created correctly.

-- You must update the partition filter to your actual business date. For example, if the task runs on 2024-08-08, the business date is 2024-08-07 (the day before the task run date).
SELECT * FROM dws_user_info_all_di_spark WHERE dt = 'your_business_date';

Edit code

Double-click the ads_user_info_1d_spark node to open the node configuration page. Enter the following SQL statements.

-- Scenario: This Spark SQL script further processes the dws_user_info_all_di_spark table by using Spark SQL functions and writes the result to a new table named ads_user_info_1d_spark.
-- Note:
--      DataWorks provides scheduling parameters to write daily incremental data to the corresponding business partition of the target table.
--      In a real-world development scenario, you can define code variables by using the ${variable_name} format and assign scheduling parameters to them on the scheduling configuration page to enable dynamic parameter passing.
CREATE TABLE IF NOT EXISTS ads_user_info_1d_spark (
  uid STRING COMMENT 'User ID',
  device STRING COMMENT 'Terminal type',
  pv BIGINT COMMENT 'PV',
  gender STRING COMMENT 'Gender',
  age_range STRING COMMENT 'Age range',
  zodiac STRING COMMENT 'Zodiac sign'
)
PARTITIONED BY (
  dt STRING
);
ALTER TABLE ads_user_info_1d_spark ADD IF NOT EXISTS PARTITION (dt='${bizdate}');
INSERT OVERWRITE TABLE ads_user_info_1d_spark PARTITION (dt='${bizdate}')
SELECT uid
  , MAX(device)
  , COUNT(0) AS pv
  , MAX(gender)
  , MAX(age_range)
  , MAX(zodiac)
FROM dws_user_info_all_di_spark
WHERE dt = '${bizdate}'
GROUP BY uid;

Configure scheduling properties

Save the configuration

Configure other required parameters as needed. When you are finished, click the icon in the toolbar of the node configuration tab to save the settings.

Verify the user profile table data

After the upstream nodes and the current node have run successfully, create an EMR Spark SQL Ad Hoc Query task to verify that the table was created correctly.

-- You must update the partition filter to your actual business date. For example, if the task runs on 2023-02-22, the business date is 2023-02-21 (the day before the task run date).
SELECT * FROM ads_user_info_1d_spark WHERE dt = 'your_business_date';