DataWorks:Process data

In the canvas of the workflow, move the pointer over the dwd_log_info_di_spark node and click Open Node.

Copy the following SQL statements and paste them in the code editor:

-- Scenario: The SQL statements in the following sample code are Spark SQL statements. You can use a Spark SQL function to split data in the ods_raw_log_d_spark table that is loaded to Spark by using ##@@ to generate multiple fields, and then synchronize the fields to the dwd_log_info_di_spark table. 
-- Note:
--      DataWorks provides scheduling parameters that you can use to write daily incremental data to the corresponding partition of the destination table in the scheduling scenario. 
--      In actual development scenarios, you can define variables in the node code in the ${Variable name} format. Then, you can assign scheduling parameters as values to the variables on the Properties tab of the configuration tab of the node. This way, the values of scheduling parameters are dynamically replaced in the node code based on the configurations of the scheduling parameters in the scheduling scenario.   
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 that is returned by the server',
  bytes STRING COMMENT 'The number of bytes that are 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)
LOCATION 'oss://dw-spark-demo.oss-cn-shanghai-internal.aliyuncs.com/dwd_log_info_di_spark/log_${bizdate}/';

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;

In the right-side navigation pane of the configuration tab of the node, click Debugging Configurations. On the Debugging Configurations tab, configure the following parameters. These parameters are used to test the workflow in Step 4.

Optional. Configure scheduling properties.

You can retain default values for parameters related to scheduling properties in this tutorial. If you want to modify the configurations of scheduling properties, you can click Properties in the right-side navigation pane of the configuration tab of the batch synchronization node. For information about other parameters on the Properties tab, see Node scheduling.

• Scheduling Parameters: In this tutorial, scheduling parameters are configured for the workflow. You do not need to configure scheduling parameters for the inner nodes of the workflow. The configured scheduling parameters can be directly used for code and tasks developed based on the inner nodes.

• Scheduling Policies: You can configure the Time for Delayed Execution parameter to specify the duration by which the running of the batch synchronization node lags behind the running of the workflow. In this tutorial, you do not need to configure this parameter.

Click Save.

In the canvas of the workflow, move the pointer over the dws_user_info_all_di_spark node and click Open Node.

Copy the following SQL statements and paste them in the code editor:

-- Scenario: The SQL statements in the following sample code are Spark SQL statements. You can join the dwd_log_info_di_spark and ods_user_info_d_spark tables based on the uid field and write data to the specified dt partition. 
-- Note:
--      DataWorks provides scheduling parameters that you can use to write daily incremental data to the corresponding partition of the destination table in the scheduling scenario. 
--      In actual development scenarios, you can define variables in the node code in the ${Variable name} format. Then, you can assign scheduling parameters as values to the variables on the Properties tab of the configuration tab of the node. This way, the values of scheduling parameters are dynamically replaced in the node code based on the configurations of the scheduling parameters in the scheduling scenario. 
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 'The terminal type',
    method     STRING COMMENT 'The HTTP request type',
    url        STRING COMMENT 'url',
    `time`     STRING COMMENT 'The time in the yyyymmddhh:mi:ss format'
)
PARTITIONED BY (dt STRING)
LOCATION 'oss://dw-spark-demo.oss-cn-shanghai-internal.aliyuncs.com/dws_user_info_all_di_spark/log_${bizdate}/';

-- 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 fact 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 (
  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;

In the right-side navigation pane of the configuration tab of the node, click Debugging Configurations. On the Debugging Configurations tab, configure the following parameters. These parameters are used to test the workflow in Step 4.

Optional. Configure scheduling properties.

You can retain default values for parameters related to scheduling properties in this tutorial. If you want to modify the configurations of scheduling properties, you can click Properties in the right-side navigation pane of the configuration tab of the batch synchronization node. For information about other parameters on the Properties tab, see Node scheduling.

• Scheduling Parameters: In this tutorial, scheduling parameters are configured for the workflow. You do not need to configure scheduling parameters for the inner nodes of the workflow. The configured scheduling parameters can be directly used for code and tasks developed based on the inner nodes.

• Scheduling Policies: You can configure the Time for Delayed Execution parameter to specify the duration by which the running of the batch synchronization node lags behind the running of the workflow. In this tutorial, you do not need to configure this parameter.

Click Save.

In the canvas of the workflow, move the pointer over the ads_user_info_1d_spark node and click Open Node.

Copy the following SQL statements and paste them in the code editor:

-- Scenario: The SQL statements in the following sample code are Spark SQL statements. You can use a Spark SQL function to further process the dws_user_info_all_di_spark table in Spark SQL and synchronize data to the ads_user_info_1d_spark table. 
-- Note:
--      DataWorks provides scheduling parameters that you can use to write daily incremental data to the corresponding partition of the destination table in the scheduling scenario. 
--      In actual development scenarios, you can define variables in the node code in the ${Variable name} format. Then, you can assign scheduling parameters as values to the variables on the Properties tab of the configuration tab of the node. This way, the values of scheduling parameters are dynamically replaced in the node code based on the configurations of the scheduling parameters in the scheduling scenario. 
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
)
LOCATION 'oss://dw-spark-demo.oss-cn-shanghai-internal.aliyuncs.com/ads_user_info_1d_spark/log_${bizdate}/';

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; 

In the right-side navigation pane of the configuration tab of the node, click Debugging Configurations. On the Debugging Configurations tab, configure the following parameters. These parameters are used to test the workflow in Step 4.

Optional. Configure scheduling properties.

You can retain default values for parameters related to scheduling properties in this tutorial. If you want to modify the configurations of scheduling properties, you can click Properties in the right-side navigation pane of the configuration tab of the batch synchronization node. For information about other parameters on the Properties tab, see Node scheduling.

• Scheduling Parameters: In this tutorial, scheduling parameters are configured for the workflow. You do not need to configure scheduling parameters for the inner nodes of the workflow. The configured scheduling parameters can be directly used for code and tasks developed based on the inner nodes.

• Scheduling Policies: You can configure the Time for Delayed Execution parameter to specify the duration by which the running of the batch synchronization node lags behind the running of the workflow. In this tutorial, you do not need to configure this parameter.

Click Save.