Use EMR Hive nodes to process OSS data for user profile analysis - DataWorks - Alibaba Cloud - DataWorks

This topic describes how to use an EMR Hive node in DataWorks to process data from the ods_user_info_d_emr user information table and the ods_raw_log_d_emr access log table. The data synchronized from OSS is processed to generate user profile data.

Prerequisite

Before you begin, complete the steps in Synchronize data.

Step 1: Design the workflow

To configure dependencies between workflow nodes, see Synchronize data.

Double-click the new workflow to open its configuration tab. Click New Node, select EMR Hive, and drag it to the canvas on the right. In the New Node dialog box, enter a node Node Name and click Confirm.

Create three EMR Hive nodes, name them dwd_log_info_di_emr, dws_user_info_all_di_emr, and ads_user_info_1d_emr, and then configure the following dependencies.

dwd_log_info_di_emr: Cleanses the raw log data from OSS.
dws_user_info_all_di_emr: Aggregates the cleansed log data and basic user information.
ads_user_info_1d_emr: Generates the final user profile data.

Step 2: Create a function

You must parse the raw log data into the target format. This topic provides a function package for translating IP addresses to regions. Download the package, register it as a function in DataWorks, and call it in your code.

Upload the resource

Download the ip2region-emr.jar file.
In DataStudio, open the WorkShop workflow, right-click EMR, and choose New Resource > EMR JAR. Configure the parameters for the new resource and click New.
The following table describes the key parameters.
- Storage Path: Select the OSS bucket that you specified for the EMR cluster when you prepared the environment.
- Upload File: Select the downloaded ip2region-emr.jar file.
Leave other parameters at their default values or configure them as needed.
Click the icon on the toolbar to submit the resource to the EMR engine project in the development environment.

Register the function

In DataStudio, open the workflow, right-click EMR, and select New Function.
In the New Function dialog box, set Function Name to getregion and click New. Then configure the function information.
The following table describes the key parameters.
- Resource: Select ip2region-emr.jar.
- Class Name: Enter org.alidata.emr.udf.Ip2Region.
Leave other parameters at their default values or configure them as needed.
Click the icon on the toolbar to submit the function to the EMR engine project in the development environment.

Step 3: Configure EMR Hive nodes

Create the dwd_log_info_di_emr node

1. Edit the code

Double-click the dwd_log_info_di_emr node to open its configuration tab. On the node editing page, enter the following statements.

Note

If multiple EMR engines are associated with your workspace in Data Studio, select the appropriate EMR Engine as needed. If only one EMR engine is associated, you do not need to select one.

--Create ODS layer table
CREATE TABLE IF NOT EXISTS dwd_log_info_di_emr (
  ip STRING COMMENT 'IP address',
  uid STRING COMMENT 'User ID',
  `time` STRING COMMENT 'Time yyyymmddhh:mi:ss',
  status STRING COMMENT 'Server response status code',
  bytes STRING COMMENT 'Bytes returned to client',
  region STRING COMMENT 'Region, derived from IP',
  method STRING COMMENT 'HTTP request type',
  url STRING COMMENT 'url',
  protocol STRING COMMENT 'HTTP protocol version',
  referer STRING COMMENT 'Source url',
  device STRING COMMENT 'Device type ',
  identity STRING COMMENT 'Access type crawler feed user 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;
INSERT OVERWRITE TABLE dwd_log_info_di_emr PARTITION (dt='${bizdate}')
SELECT ip
  , uid
  , tm
  , status
  , bytes 
  , getregion(ip) AS region --Use custom UDF to derive region from IP. 
  , regexp_extract(request, '(^[^ ]+) .*') AS method --Split request into three fields via regex.
  , regexp_extract(request, '^[^ ]+ (.*) [^ ]+$') AS url
  , regexp_extract(request, '.* ([^ ]+$)') AS protocol 
  , regexp_extract(referer, '^[^/]+://([^/]+){1}') AS referer  --Clean referer via regex to get a more precise url.
  , CASE
    WHEN lower(agent) RLIKE 'android' THEN 'android' --Derive device type and access type from agent.
    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, '^[^ ]+ (.*) [^ ]+$') 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
  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_emr
  WHERE dt = '${bizdate}'
) a;

2. Configure schedule settings

The following schedule settings are used to implement the scheduling scenario: Every day at 00:30, after the upstream ods_raw_log_d_emr node synchronizes the user_log.txt data from OSS to the ods_raw_log_d_emr table in EMR, the current dwd_log_info_di_emr node is triggered to process the data in the ods_raw_log_d_emr table and write the results to the corresponding business date partition of the dwd_log_info_di_emr table.

Parameter	Configuration
Scheduling Parameters	In the Scheduling Parameters section, add the following: Parameter name: bizdate Parameter value: $[yyyymmdd-1]
Time attribute	Set RUN Attribute to Allow Rerun on Success or Failure.
Scheduling Dependency	In Scheduling Dependency, verify that the output table is configured as the output of this node. The format is `WorkSpaceName.NodeName`.

Note

For the time settings, set Scheduling period to Day. You do not need to separately configure the Scheduling Time for this node. The daily trigger time of this node is controlled by the scheduled time of the workflow virtual node Workshop_start_emr, which means the node is not scheduled until after 00:30 each day.

3. Save the configuration

Configure other required parameters as needed. After the configuration is complete, click the icon on the toolbar of the node code editing page to save the current configuration.

Create the dws_user_info_all_di_emr node

1. Edit the code

Double-click the dws_user_info_all_di_emr node to open its configuration tab. On the node editing page, enter the following statements.

Note

If multiple EMR engines are associated with your workspace in Data Studio, select the appropriate EMR Engine as needed. If only one EMR engine is associated, you do not need to select one.

--Create DW layer table
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',
  device STRING COMMENT 'Device type ',
  identity STRING COMMENT 'Access type crawler feed user unknown',
  method STRING COMMENT 'HTTP request type',
  url STRING COMMENT 'url',
  referer STRING COMMENT 'Source url',
  `time` STRING COMMENT 'Time yyyymmddhh:mi:ss'
)
PARTITIONED BY (
  dt STRING
);

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

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;

2. Configure schedule settings

The following schedule settings are used to implement the scheduling scenario: Every day at 00:30, after the upstream tasks ods_user_info_d_emr and dwd_log_info_di_emr are complete, the dws_user_info_all_di_emr node is triggered to merge and process data from the ods_user_info_d_emr and dwd_log_info_di_emr tables and write the results to the dws_user_info_all_di_emr table.

Parameter	Configuration
Scheduling Parameters	In the Scheduling Parameters section, add the following: Parameter name: bizdate Parameter value: $[yyyymmdd-1]
Time attribute	Set RUN Attribute to Allow Rerun on Success or Failure.
Scheduling Dependency	In Scheduling Dependency, verify that the output table is configured as the output of this node. The format is `WorkSpaceName.NodeName`.

Note

3. Save the configuration

Configure other required parameters as needed. After the configuration is complete, click the icon on the toolbar of the node code editing page to save the current configuration.

Create the ads_user_info_1d_emr node

1. Edit the code

Double-click the ads_user_info_1d_emr node to open its configuration tab. On the node editing page, enter the following statements.

Note

If multiple EMR engines are associated with your workspace in Data Studio, select the appropriate EMR Engine as needed. If only one EMR engine is associated, you do not need to select one.

--Create RPT layer table
CREATE TABLE IF NOT EXISTS ads_user_info_1d_emr (
  uid STRING COMMENT 'User ID',
  region STRING COMMENT 'Region, derived from IP',
  device STRING COMMENT 'Device type ',
  pv BIGINT COMMENT 'pv',
  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}');

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;

2. Configure schedule settings

After the upstream dws_user_info_all_di_emr node finishes merging the ods_user_info_d_emr and dwd_log_info_di_emr tables, the ads_user_info_1d_emr node is triggered to further process the data and generate consumable data.

Parameter	Configuration
Scheduling Parameters	In the Scheduling Parameters section, add the following: Parameter name: bizdate Parameter value: $[yyyymmdd-1]
Time attribute	Set RUN Attribute to Allow Rerun on Success or Failure.
Scheduling Dependency	In Scheduling Dependency, verify that the output table is configured as the output of this node. The format is `WorkSpaceName.NodeName`.

Note

3. Save the configuration

Configure other required parameters as needed. After the configuration is complete, click the icon on the toolbar of the node code editing page to save the current configuration.

Step 4: Submit the workflow

After you complete all configurations for the workflow, test whether the workflow runs as expected. After a successful test, submit the workflow and wait for it to be deployed.

On the workflow editing page, click to run the workflow.
After appears next to all nodes in the workflow, click to submit the workflow that runs successfully.
In the Submit dialog box, select the nodes to submit, select Ignore I/O Inconsistency Alerts, and then click Confirm.
After the submission is successful, deploy each workflow node.
1. Click Publish on the right side of the page to go to the Create Deployment Package page.
2. Select the nodes to deploy, click Deploy, and in the Confirm Release dialog box, click Publish.

Step 5: Run the task in the production environment

After you deploy a task, instances are generated and run the next day. You can use Supplementary data to backfill data for the deployed workflow and verify that the task runs as expected in the production environment. For more information, see Backfill data.

After the task is deployed, click Operation and Maintenance Center in the upper-right corner.
You can also open the workflow editing page and click Go to operations on the toolbar to go to the Operation and Maintenance Center page.
In the left-side navigation pane, choose Auto Triggered Task O&M > Auto Triggered Node to go to the Auto Triggered Node page, and click the workshop_start_emr virtual node.
In the DAG on the right, right-click the workshop_start_emr node and choose Supplementary data > Current and Descendant Nodes Retroactively.
Select the tasks for which you want to backfill data, enter the business date, and click Submit and Redirect.
On the backfill data page, click Refresh until all SQL tasks have run successfully.

Next step

In periodic scheduling scenarios, to ensure that the data generated by tasks meets expectations, you can configure data quality monitoring for the output tables. For more information, see Monitor data quality.