DataWorks:Process data

Configure the dwd_log_info_di_spark node

This node splits the raw log table ods_raw_log_d_spark — where each row is a single delimited string — into structured fields, and writes the result to dwd_log_info_di_spark.

Step 1: Configure the node code

Double-click the dwd_log_info_di_spark node to open its configuration tab. Enter the following SQL:

-- Split the ods_raw_log_d_spark table into structured fields and write to dwd_log_info_di_spark.
-- The source table uses ##@@ as the field delimiter within a single column (col).
-- DataWorks scheduling parameter ${bizdate} is replaced at runtime with the business date (T-1).

CREATE TABLE IF NOT EXISTS dwd_log_info_di_spark (
  ip       STRING COMMENT 'The IP address',
  uid      STRING COMMENT 'The user ID',
  tm       STRING COMMENT 'The time in the yyyymmddhh:mi:ss format',
  status   STRING COMMENT 'The status code returned by the server',
  bytes    STRING COMMENT 'The number of bytes returned to the client',
  method   STRING COMMENT 'The request method',
  url      STRING COMMENT 'url',
  protocol STRING COMMENT 'The 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;

Step 2: Configure scheduling properties

On the Properties tab of the node configuration page:

• In the Scheduling Parameter section, click Add Parameter, set Parameter Name to bizdate, and set Parameter Value to $[yyyymmdd-1].

• In the Dependencies section, confirm that the output name follows the workspacename.nodename format. For details, see Configure scheduling dependencies.

In the Schedule section, set Scheduling Cycle to Day. The scheduled run time is inherited from the workshop_start_spark zero load node, which acts as the trigger for all downstream nodes. This node runs after 00:30 every day without additional time configuration.

Step 3: (Optional) Configure advanced Spark parameters

On the Advanced Settings tab in the right-side navigation pane, configure the following Spark system parameters as needed.

To add a custom Spark parameter, enter it on the Advanced Settings tab. For example, specifying spark.eventLog.enabled: false causes DataWorks to deliver it to the EMR Serverless Spark workspace as --conf spark.eventLog.enabled=false. To configure parameters globally, see Configure global Spark parameters. For all available parameters, see Spark Configuration.

Step 4: Save

Click the save icon in the top toolbar to save the node configuration.

Step 5: Verify the output

After the ancestor nodes and this node run successfully, verify that the split log table was created correctly.

In the left-side navigation pane of DataStudio, click Ad Hoc Query. Create an ad hoc query task of the EMR Spark SQL type and run the following query, replacing <data_timestamp> with the actual data timestamp (T-1):

SELECT * FROM dwd_log_info_di_spark WHERE dt = '<data_timestamp>';

Configure the dws_user_info_all_di_spark node

This node joins the structured log table dwd_log_info_di_spark with the user information table ods_user_info_d_spark on the uid field, producing the aggregate table dws_user_info_all_di_spark.

Step 1: Configure the node code

Double-click the dws_user_info_all_di_spark node to open its configuration tab. Enter the following SQL:

-- Join dwd_log_info_di_spark and ods_user_info_d_spark on uid and write to dws_user_info_all_di_spark.
-- DataWorks scheduling parameter ${bizdate} is replaced at runtime with the business date (T-1).

CREATE TABLE IF NOT EXISTS dws_user_info_all_di_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',
    device    STRING COMMENT 'Terminal type',
    method    STRING COMMENT 'HTTP request type',
    url       STRING COMMENT 'url',
    `time`    STRING COMMENT 'The time in the 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}');

-- Join user information and log data and insert into the partition.
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 dwd_log_info_di_spark AS a
LEFT OUTER JOIN ods_user_info_d_spark AS b
ON a.uid = b.uid;

Step 2: Configure scheduling properties

Follow the same steps as for the dwd_log_info_di_spark node: set bizdate to $[yyyymmdd-1], confirm the output name in Dependencies, and set Scheduling Cycle to Day. This node also runs after 00:30 every day, inheriting its scheduled time from the workshop_start_spark zero load node.

Step 3: (Optional) Configure advanced Spark parameters

Follow the same steps as for the dwd_log_info_di_spark node. The available parameters (SERVERLESS_RELEASE_VERSION, SERVERLESS_QUEUE_NAME, SERVERLESS_SQL_COMPUTE, FLOW_SKIP_SQL_ANALYZE) are the same across all three nodes.

Step 4: Save

Click the save icon in the top toolbar to save the node configuration.

Step 5: Verify the output

After the ancestor nodes and this node run successfully, verify that the joined table was created correctly.

In the left-side navigation pane of DataStudio, click Ad Hoc Query. Create an ad hoc query task of the EMR Spark SQL type and run the following query, replacing <data_timestamp> with the actual data timestamp (T-1):

SELECT * FROM dws_user_info_all_di_spark WHERE dt = '<data_timestamp>';

Configure the ads_user_info_1d_spark node

This node aggregates the joined table dws_user_info_all_di_spark using MAX and COUNT operations, producing ads_user_info_1d_spark as the final user profile table.

Step 1: Configure the node code

Double-click the ads_user_info_1d_spark node to open its configuration tab. Enter the following SQL:

-- Aggregate dws_user_info_all_di_spark to generate the user profile table ads_user_info_1d_spark.
-- DataWorks scheduling parameter ${bizdate} is replaced at runtime with the business date (T-1).

CREATE TABLE IF NOT EXISTS ads_user_info_1d_spark (
  uid       STRING COMMENT 'The user ID',
  device    STRING COMMENT 'The terminal type',
  pv        BIGINT COMMENT 'Page views (PV)',
  gender    STRING COMMENT 'The gender',
  age_range STRING COMMENT 'The age range',
  zodiac    STRING COMMENT 'The 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;

Step 2: Configure scheduling properties

Follow the same steps as for the previous nodes: set bizdate to $[yyyymmdd-1], confirm the output name in Dependencies, and set Scheduling Cycle to Day.

Step 3: (Optional) Configure advanced Spark parameters

Follow the same steps as for the dwd_log_info_di_spark node.

Step 4: Save

Click the save icon in the top toolbar to save the node configuration.

Step 5: Verify the output

After the ancestor nodes and this node run successfully, verify that the user profile table was created correctly.

In the left-side navigation pane of DataStudio, click Ad Hoc Query. Create an ad hoc query task of the EMR Spark SQL type and run the following query, replacing <data_timestamp> with the actual data timestamp (T-1):

SELECT * FROM ads_user_info_1d_spark WHERE dt = '<data_timestamp>';