E-MapReduce:Processar dados

-- Create the DWD-layer table for cleansed log data.
CREATE TABLE IF NOT EXISTS dwd_log_info_di_emr (
  ip       STRING COMMENT 'Visitor IP address',
  uid      STRING COMMENT 'User ID',
  `time`   STRING COMMENT 'Request timestamp (yyyymmddhh:mi:ss)',
  status   STRING COMMENT 'HTTP status code returned by the server',
  bytes    STRING COMMENT 'Bytes returned to the client',
  region   STRING COMMENT 'Region derived from the IP address via UDF',
  method   STRING COMMENT 'HTTP method (GET, POST, etc.)',
  url      STRING COMMENT 'Request URL path',
  protocol STRING COMMENT 'HTTP version',
  referer  STRING COMMENT 'Referring URL (domain only)',
  device   STRING COMMENT 'Device type: android, iphone, ipad, macintosh, windows_phone, windows_pc, or unknown',
  identity STRING COMMENT 'Access type: crawler, feed, user, or unknown'
)
PARTITIONED BY (dt STRING);

ALTER TABLE dwd_log_info_di_emr ADD IF NOT EXISTS PARTITION (dt='${bizdate}');

SET hive.vectorized.execution.enabled = false;

-- Parse and cleanse raw log data for the current business date.
-- Each raw row is a single string with fields separated by ##@@.
-- Field order: [0]=ip, [1]=uid, [2]=tm, [3]=request, [4]=status, [5]=bytes, [6]=referer, [7]=agent
INSERT OVERWRITE TABLE dwd_log_info_di_emr PARTITION (dt='${bizdate}')
SELECT
    ip
  , uid
  , tm
  , status
  , bytes
  , getregion(ip)                                            AS region   -- UDF: translates IP to region name
  , regexp_substr(request, '(^[^ ]+ )')                     AS method   -- Extracts HTTP method from request string
  , regexp_extract(request, '^[^ ]+ (.*) [^ ]+$')           AS url      -- Extracts URL path from request string
  , regexp_extract(request, '.* ([^ ]+$)')                   AS protocol -- Extracts HTTP version from request string
  , regexp_extract(referer, '^[^/]+://([^/]+){1}')           AS referer  -- Strips path; keeps only the referring domain
  , 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   -- Derived from the User-Agent string
  , CASE
      WHEN lower(agent) RLIKE '(bot|spider|crawler|slurp)'                              THEN 'crawler'
      WHEN lower(agent) RLIKE 'feed'
        OR regexp_extract(request, '^[^ ]+ (.*) [^ ]+$') RLIKE 'feed'                  THEN 'feed'
      WHEN lower(agent) NOT RLIKE '(bot|spider|crawler|feed|slurp)'
        AND agent RLIKE '^[Mozilla|Opera]'
        AND regexp_extract(request, '^[^ ]+ (.*) [^ ]+$') NOT RLIKE 'feed'             THEN 'user'
      ELSE 'unknown'
    END AS identity -- Derived from the User-Agent string and URL path
FROM (
  -- Split each raw log row into individual fields using the ##@@ delimiter.
  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_emr
  WHERE dt = '${bizdate}'
) a;

-- Create the DWS-layer table combining log behavior and user demographics.
CREATE TABLE IF NOT EXISTS dws_user_info_all_di_emr (
  uid       STRING COMMENT 'User ID',
  gender    STRING COMMENT 'Gender',
  age_range STRING COMMENT 'Age range',
  zodiac    STRING COMMENT 'Zodiac sign',
  region    STRING COMMENT 'Region derived from IP address',
  device    STRING COMMENT 'Device type',
  identity  STRING COMMENT 'Access type: crawler, feed, user, or unknown',
  method    STRING COMMENT 'HTTP method',
  url       STRING COMMENT 'Request URL path',
  referer   STRING COMMENT 'Referring domain',
  `time`    STRING COMMENT 'Request timestamp (yyyymmddhh:mi:ss)'
)
PARTITIONED BY (dt STRING);

ALTER TABLE dws_user_info_all_di_emr ADD IF NOT EXISTS PARTITION (dt='${bizdate}');

-- Join cleansed log data (left) with user demographics (right) on uid.
-- COALESCE handles cases where a uid appears in logs but not in the user table.
INSERT OVERWRITE TABLE dws_user_info_all_di_emr PARTITION (dt='${bizdate}')
SELECT
    COALESCE(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`
FROM (
  SELECT * FROM dwd_log_info_di_emr WHERE dt = '${bizdate}'
) a
LEFT OUTER JOIN (
  SELECT * FROM ods_user_info_d_emr WHERE dt = '${bizdate}'
) b ON a.uid = b.uid;

-- Create the ADS-layer table for daily user profiles.
CREATE TABLE IF NOT EXISTS ads_user_info_1d_emr (
  uid       STRING COMMENT 'User ID',
  region    STRING COMMENT 'Region derived from IP address',
  device    STRING COMMENT 'Device type',
  pv        BIGINT COMMENT 'Page view count',
  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_emr ADD IF NOT EXISTS PARTITION (dt='${bizdate}');

-- Aggregate per-user behavior for the current business date.
-- COUNT(0) counts all log rows per uid as the page view total.
INSERT OVERWRITE TABLE ads_user_info_1d_emr PARTITION (dt='${bizdate}')
SELECT
    uid
  , MAX(region)
  , MAX(device)
  , COUNT(0)    AS pv
  , MAX(gender)
  , MAX(age_range)
  , MAX(zodiac)
FROM dws_user_info_all_di_emr
WHERE dt = '${bizdate}'
GROUP BY uid;