加工数据

配置代码

双击dwd_log_info_di_spark节点，进入节点配置页面，编写如下语句。

-- 场景：以下SQL为Spark SQL，通过Spark SQL函数将加载至Spark中的ods_raw_log_d_spark按"##@@"进行切分后生成多个字段，并写入新表dwd_log_info_di_spark。
-- 补充：
--      DataWorks提供调度参数，可实现调度场景下，将每日增量数据写入目标表对应的业务分区内。
--      在实际开发场景下，您可通过${变量名}格式定义代码变量，并在调度配置页面通过变量赋值调度参数的方式，实现调度场景下代码动态入参。  

CREATE TABLE IF NOT EXISTS dwd_log_info_di_spark (
  ip STRING COMMENT 'ip地址',
  uid STRING COMMENT '用户ID',
  tm STRING COMMENT '时间yyyymmddhh:mi:ss',
  status STRING COMMENT '服务器返回状态码',
  bytes STRING COMMENT '返回给客户端的字节数',
  method STRING COMMENT'请求方法',
  url STRING COMMENT 'url',
  protocol STRING COMMENT '协议',
  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;

配置调度属性

保存配置

本案例其他必填配置项，您可按需自行配置，配置完成后，在节点代码编辑页面，单击工具栏中的按钮，保存当前配置。

验证日志表拆分情况

在确保上游节点以及本节点运行成功的情况下，在左侧导航栏的临时查询中新建EMR Spark SQL临时查询，编写SQL查看EMR Spark SQL节点创建的表是否正常产出。

-- 您需要将分区过滤条件更新为您当前操作的实际业务日期。例如，任务运行的日期为20230222，则业务日期为20230221，即任务运行日期的前一天。
SELECT * FROM dwd_log_info_di_spark WHERE dt ='业务日期';

编辑代码

双击dws_user_info_all_di_spark节点，进入节点配置页面。在节点编辑页面，编写如下语句。

-- 场景：以下SQL为Spark SQL，通过uid将dwd_log_info_di_spark和ods_user_info_d_spark进行关联，并写入对应的dt分区。
-- 补充：
--      DataWorks提供调度参数，可实现调度场景下，将每日增量数据写入目标表对应的业务分区内。
--      在实际开发场景下，您可通过${变量名}格式定义代码变量，并在调度配置页面通过变量赋值调度参数的方式，实现调度场景下代码动态入参。
CREATE TABLE IF NOT EXISTS dws_user_info_all_di_spark (
    uid        STRING COMMENT '用户ID',
    gender     STRING COMMENT '性别',
    age_range  STRING COMMENT '年龄段',
    zodiac     STRING COMMENT '星座',
    device     STRING COMMENT '终端类型 ',
    method     STRING COMMENT 'http请求类型',
    url        STRING COMMENT 'url',
    `time`     STRING COMMENT '时间yyyymmddhh:mi:ss'
)
PARTITIONED BY (dt STRING);

--添加分区
ALTER TABLE dws_user_info_all_di_spark ADD IF NOT EXISTS PARTITION (dt = '${bizdate}');

--插入user表与日志表数据
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;

配置调度属性

保存配置

本案例其他必填配置项，您可按需自行配置，配置完成后，在节点代码编辑页面，单击工具栏中的按钮，保存当前配置。

验证用户日志明细表数据情况

在确保上游节点以及本节点运行成功的情况下，在左侧导航栏的临时查询中新建EMR Spark SQL临时查询，编写SQL查看EMR Spark SQL节点创建的表是否正常产出。

-- 您需要将分区过滤条件更新为您当前操作的实际业务日期。例如，任务运行的日期为20240808，则业务日期为20240807，即任务运行日期的前一天。
SELECT * FROM dws_user_info_all_di_spark WHERE dt ='业务日期';

编辑代码

双击ads_user_info_1d_spark节点，进入节点配置页面。在节点编辑页面，编写如下语句。

-- 场景：以下SQL为Spark SQL，通过Spark SQL函数将Spark中的dws_user_info_all_di_spark表进一步的加工，并写入新表ads_user_info_1d_spark。
-- 补充：
--      DataWorks提供调度参数，可实现调度场景下，将每日增量数据写入目标表对应的业务分区内。
--      在实际开发场景下，您可通过${变量名}格式定义代码变量，并在调度配置页面通过变量赋值调度参数的方式，实现调度场景下代码动态入参。
CREATE TABLE IF NOT EXISTS ads_user_info_1d_spark (
  uid STRING COMMENT '用户ID',
  device STRING COMMENT '终端类型 ',
  pv BIGINT COMMENT 'pv',
  gender STRING COMMENT '性别',
  age_range STRING COMMENT '年龄段',
  zodiac STRING COMMENT '星座'
)
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;

调度配置

保存配置

本案例其他必填配置项，您可按需自行配置，配置完成后，在节点代码编辑页面，单击工具栏中的按钮，保存当前配置。

验证用户画像表数据情况

在确保上游节点以及本节点运行成功的情况下，在左侧导航栏的临时查询中新建EMR Spark SQL临时查询，编写SQL查看该节点创建的表是否正常产出。

-- 您需要将分区过滤条件更新为您当前操作的实际业务日期。例如，任务运行的日期为20230222，则业务日期为20230221，即任务运行日期的前一天。
SELECT * FROM ads_user_info_1d_spark WHERE dt ='业务日期';