E-MapReduce:Process data

1. Configure the node code

Double-click the dwd_log_info_di_spark node to open its configuration tab, and enter the following SQL:

-- Splits ods_raw_log_d_spark using ##@@ as a delimiter
-- and writes the structured fields to dwd_log_info_di_spark.
-- ${bizdate} is a scheduling parameter replaced at runtime with the data timestamp (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;

2. Configure scheduling properties

Set Scheduling Cycle to Day. The scheduled run time is determined by the workshop_start_spark zero load node — all nodes in this workflow run after 00:30 daily.

4. Save the configuration

Click the icon in the top toolbar to save.

5. Verify the split result

After the ancestor nodes and this node run successfully, click Ad Hoc Query in the left-side navigation pane of the DataStudio page. Create an ad hoc query task of the EMR Spark SQL type and run the following query to confirm that the log fields are parsed correctly. Check that method, url, protocol, device, and identity contain expected values:

-- Replace '20240807' with the actual bizdate value for your run.
-- The data timestamp is one day earlier than the scheduling date.
SELECT ip, uid, method, url, device, identity
FROM dwd_log_info_di_spark
WHERE dt = '20240807'
LIMIT 20;

In batch computing scenarios, bizdate represents the business date — the day on which the data was generated, not the day the node runs. For example, if the node runs on August 8, 2024, the data timestamp is 20240807.

1. Configure the node code

Double-click the dws_user_info_all_di_spark node to open its configuration tab, and enter the following SQL:

-- Joins the clean log table and the user information table on uid,
-- then writes the merged data to dws_user_info_all_di_spark.
-- ${bizdate} is replaced at runtime with the data timestamp (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}');

-- Insert data from the basic user information table and the new log table.
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;

2. Configure scheduling properties

Set Scheduling Cycle to Day. The scheduled run time is determined by the workshop_start_spark zero load node.

4. Save the configuration

Click the icon in the top toolbar to save.

5. Verify the join result

After the ancestor nodes and this node run successfully, click Ad Hoc Query in the DataStudio page. Create an EMR Spark SQL ad hoc query task and run the following query. Confirm that gender, age_range, and zodiac are populated for rows where a matching uid exists in ods_user_info_d_spark, and that the total row count matches the row count in dwd_log_info_di_spark for the same partition:

-- Replace '20240807' with the actual bizdate value for your run.
SELECT uid, gender, age_range, zodiac, device, method
FROM dws_user_info_all_di_spark
WHERE dt = '20240807'
LIMIT 20;

Because this is a left join, rows in the log table without a matching uid in the user information table will have NULL values for gender, age_range, and zodiac.

1. Configure the node code

Double-click the ads_user_info_1d_spark node to open its configuration tab, and enter the following SQL:

-- Aggregates dws_user_info_all_di_spark by uid,
-- computing pv (page views) and representative user attributes.
-- ${bizdate} is replaced at runtime with the data timestamp (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 '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;

2. Configure scheduling properties

Set Scheduling Cycle to Day. The scheduled run time is determined by the workshop_start_spark zero load node.

3. (Optional) Configure Spark system parameters

Configure the same Spark system parameters described in the dwd_log_info_di_spark node's step 3.

4. Save the configuration

Click the icon in the top toolbar to save.

5. Verify the user profile table

After the ancestor nodes and this node run successfully, click Ad Hoc Query in the DataStudio page. Create an EMR Spark SQL ad hoc query task and run the following query. Confirm that each user has exactly one row, pv is greater than zero, and profile attributes such as gender and device reflect the expected distribution:

-- Replace '20240807' with the actual bizdate value for your run.
SELECT uid, device, pv, gender, age_range, zodiac
FROM ads_user_info_1d_spark
WHERE dt = '20240807'
ORDER BY pv DESC
LIMIT 20;

pv equals COUNT(0) grouped by uid, representing the number of log records for that user on the data timestamp.