通過EMR Hive節點加工資料構建使用者Portrait analysis案例-巨量資料開發治理平台 DataWorks-阿里雲

本文為您介紹如何使用DataWorks中的EMR Hive節點，對同步至OSS的使用者資訊表（ods_user_info_d_emr）及訪問日誌資料表（ods_raw_log_d_emr）中的資料進行加工，進而得到目標使用者畫像資料。

前提條件

開始本案例前，您需完成同步資料中的操作。

一、搭建資料加工鏈路

雙擊開啟您在同步資料中建立的工作流程workshop_emr，進入工作流程編排頁面，繼續添加工作流程節點。

此處需要建立3個EMR Hive節點，依次命名為dwd_log_info_di_emr、dws_user_info_all_di_emr和ads_user_info_1d_emr，並配置如下圖所示的依賴關係。

本教程節點名稱樣本及作用如下：

節點類型	節點名稱	節點作用
EMR Hive節點	dwd_log_info_di_emr	用於對原始OSS日誌資料進行清洗，使用內建函數、自訂函數（`getregion`）等完成原始日誌`ods_raw_log_d_emr`資料拆分寫入`dwd_log_info_di_emr`表多個欄位。
EMR Hive節點	dws_user_info_all_di_emr	用於將清洗後的日誌資料和使用者基本資料資料進行匯總，對使用者基本資料表（`ods_user_info_d_emr`）和初步加工後的日誌資料表（`dwd_log_info_di_emr`）進行匯總，將資料寫入`dws_user_info_all_di_emr`表中。
EMR Hive節點	ads_user_info_1d_emr	對`dws_user_info_all_di_emr`表中資料進一步加工，將資料寫入`ads_user_info_1d_emr`表，用於產生終端使用者畫像資料。

二、註冊自訂函數

為了後續資料處理任務的順利進行，您需要註冊EMR自訂函數（getregion），將同步資料階段同步至EMR的日誌資料結構拆解成表格。

上傳EMR Jar資源（ip2region.jar）

下載資源JAR包。
下載ip2region-emr.jar。

建立EMR Jar資源。

進入DataWorks工作空間列表頁，在頂部切換至目標地區，找到目標工作空間，單擊操作列的快速進入 > Data Studio，進入Data Studio。
在左側導覽列單擊，進入資源管理。
在資源管理頁面，單擊建立，選擇EMR Jar資源類型，自訂資源名。

進入資源編輯頁配置EMR Jar資源資訊，關鍵參數配置如下：

參數	說明
檔案來源	選擇本地。
檔案內容	點擊上傳下載的資源JAR包`ip2region-emr.jar`。
儲存路徑	選擇OSS。選擇準備環境中建立EMR叢集配置的OSS Bucket。
資料來源	選擇您在同步資料中綁定的計算資源。
資源群組	選擇您在準備環境中建立的Serverless資源群組。

單擊儲存，並發布資源至開發、生產環境。

註冊EMR函數（getregion）

建立函數。
右鍵您所建立的EMR Jar資源，選擇建立函數... > EMR Function，自訂函數名為getregion，單擊確認進行建立。

註冊函數。

進入建立函數配置頁配置EMR Function函數資訊，關鍵參數配置如下：

參數	說明
函數類型	選擇OTHER。
資料來源	選擇您在同步資料中綁定的計算資源。
EMR資料庫	選擇default。
資源群組	選擇您在準備環境中建立的Serverless資源群組。
責任人	您可在此選擇有許可權的責任人。
類名	`org.alidata.emr.udf.Ip2Region`。
資源清單	選擇您建立的EMR Jar資源名稱。

發布函數。
單擊儲存，並發布函數至開發、生產環境。

三、配置EMR節點

資料加工需要將每層加工邏輯通過EMR Hive調度實現，本教程已提供完整的資料加工SQL範例程式碼，您需要依次為dwd_log_info_di_emr、dws_user_info_all_di_emr和ads_user_info_1d_emr節點配置。

配置dwd_log_info_di_emr節點

節點內容編輯。

在工作流程編排頁面，滑鼠懸浮於EMR Hive節點dwd_log_info_di_emr上，單擊上方出現的開啟節點，在溫馨提示彈框中選擇儲存並開啟，進入EMR Hive節點編輯頁面。輸入以下樣本SQL語句。

--建立ODS層表
CREATE TABLE IF NOT EXISTS dwd_log_info_di_emr (
  ip STRING COMMENT 'ip地址',
  uid STRING COMMENT '使用者ID',
  `time` STRING COMMENT '時間yyyymmddhh:mi:ss',
  status STRING COMMENT '伺服器返回狀態代碼',
  bytes STRING COMMENT '返回給用戶端的位元組數',
  region STRING COMMENT '地區，根據ip得到',
  method STRING COMMENT 'http請求類型',
  url STRING COMMENT 'url',
  protocol STRING COMMENT 'http協議版本號碼',
  referer STRING COMMENT '來源url',
  device STRING COMMENT '終端類型 ',
  identity STRING COMMENT '訪問類型 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 --使用自訂UDF通過ip得到地區。 
  , regexp_extract(request, '(^[^ ]+) .*') AS method --通過正則把request差分為三個欄位。
  , regexp_extract(request, '^[^ ]+ (.*) [^ ]+$') AS url
  , regexp_extract(request, '.* ([^ ]+$)') AS protocol 
  , regexp_extract(referer, '^[^/]+://([^/]+){1}') AS referer  --通過正則清洗refer，得到更精準的url。
  , CASE
    WHEN lower(agent) RLIKE 'android' THEN 'android' --通過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;

配置調試參數。

在EMR Hive節點編輯頁面右側單擊回合組態，配置以下參數，用於在步驟四調試運行中使用回合組態的相關參數測試回合。

配置項	配置說明
計算資源	選擇準備環境階段綁定的EMR計算資源。
資源群組	選擇準備環境階段購買的Serverless資源群組。
指令碼參數	無需配置。本教程提供的範例程式碼中統一使用`${bizdate}`表示業務日期，在步驟四調試運行工作流程時，設定本次運行值為具體常量（例如`20250223`），任務運行將會使用此常量替換任務中定義的變數。

配置完成後，單擊工具列中的表徵圖儲存任務節點。

配置dws_user_info_all_di_emr節點

節點內容編輯。

在工作流程編排頁面，滑鼠懸浮於EMR Hive節點dws_user_info_all_di_emr上，單擊上方出現的開啟節點，在溫馨提示彈框中選擇儲存並開啟，進入EMR Hive節點編輯頁面。輸入以下樣本SQL語句。

--建立DW層表
CREATE TABLE IF NOT EXISTS dws_user_info_all_di_emr (
  uid STRING COMMENT '使用者ID',
  gender STRING COMMENT '性別',
  age_range STRING COMMENT '年齡段',
  zodiac STRING COMMENT '星座',
  region STRING COMMENT '地區，根據ip得到',
  device STRING COMMENT '終端類型 ',
  identity STRING COMMENT '訪問類型 crawler feed user unknown',
  method STRING COMMENT 'http請求類型',
  url STRING COMMENT 'url',
  referer STRING COMMENT '來源url',
  `time` STRING COMMENT '時間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;

配置調試參數。

在EMR Hive節點編輯頁面右側單擊回合組態，配置以下參數，用於在步驟四調試運行中使用回合組態的相關參數測試回合。

配置項	配置說明
計算資源	選擇準備環境階段綁定的EMR計算資源。
資源群組	選擇準備環境階段購買的Serverless資源群組。
指令碼參數	無需配置。本教程提供的範例程式碼中統一使用`${bizdate}`表示業務日期，在步驟四調試運行工作流程時，設定本次運行值為具體常量（例如`20250223`），任務運行將會使用此常量替換任務中定義的變數。

配置完成後，單擊工具列中的表徵圖儲存任務節點。

配置ads_user_info_1d_emr節點

節點內容編輯。

在工作流程編排頁面，滑鼠懸浮於EMR Hive節點ads_user_info_1d_emr上，單擊上方出現的開啟節點，在溫馨提示彈框中選擇儲存並開啟，進入EMR Hive節點編輯頁面。輸入以下樣本SQL語句。

--建立RPT層表
CREATE TABLE IF NOT EXISTS ads_user_info_1d_emr (
  uid STRING COMMENT '使用者ID',
  region STRING COMMENT '地區，根據ip得到',
  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_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;

配置調試參數。

在EMR Hive節點編輯頁面右側單擊回合組態，配置以下參數，用於在步驟四調試運行中使用回合組態的相關參數測試回合。

配置項	配置說明
計算資源	選擇準備環境階段綁定的EMR計算資源。
資源群組	選擇準備環境階段購買的Serverless資源群組。
指令碼參數	無需配置。本教程提供的範例程式碼中統一使用`${bizdate}`表示業務日期，在步驟四調試運行工作流程時，設定本次運行值為具體常量（例如`20250223`），任務運行將會使用此常量替換任務中定義的變數。

配置完成後，單擊工具列中的表徵圖儲存任務節點。

四、加工資料

加工資料。
在Workflow畫布頂部工具列中，單擊運行，設定各節點定義的參數變數在本次運行中的取值（本教程使用20250223，您可以按需修改），單擊確定後，等待運行完成。
查詢資料加工結果。
1. 進入DataWorks工作空間列表頁，在頂部切換至目標地區，找到目標工作空間，單擊操作列的快速進入 > Data Studio，進入Data Studio。
2. 在Data Studio頁面左側導覽列單擊按鈕，進入資料開發頁面後，在左側分類樹中選擇專案目錄。
3. 按右鍵您所建立的work目錄，選擇建立節點... > EMR > EMR Hive，自訂EMR Hive節點名，單擊確認進行建立。
4. 進入EMR Hive節點編輯頁面，將下面指令碼中的業務日期替換為當前業務日期，確認同步資料寫入結果。查看匯入ods_raw_log_d_emr和ods_user_info_d_emr的記錄數。
  說明
  查詢語句中的分區列dt需要更新為實際的業務日期。例如，任務啟動並執行日期為20250223，則業務日期為20250222，即任務運行日期的前一天。
```
SELECT * FROM ads_user_info_1d_emr WHERE dt=業務日期; 
```
  - 上述命令查詢存在資料，即表示資料加工已完成。
  - 如果沒有資料，請確保運行工作流程時，配置的本次運行值與此處查詢時dt指定的業務日期一致，您可以單擊工作流程，單擊右側的運行歷史，在運行記錄右側操作列單擊查看，然後在工作流程的作業記錄中確認運行工作流程時業務日期的取值（partition=[pt=xxx]）。

五、發布工作流程

任務需要發布至生產環境後才可自動調度運行，您可以參考如下步驟，將工作流程發布至生產環境。

說明

本教程已在工作流程調度配置中統一配置了調度參數，發布前無需再為每個節點單獨配置調度參數。

在Data Studio左側導覽列單擊，然後在專案目錄地區找到已建立好的工作流程，單擊進入工作流程看板。
單擊頁面頂部工具列中的發布，開啟發布面板。
單擊開始發布生產，根據發布流程引導，完成發布。

六、在生產環境運行任務

任務發布後，在次日才會產生執行個體運行，您可以通過補資料來對發行流程進行補資料操作，以便查看任務在生產環境是否可以運行，詳情可參見補資料執行個體營運。

任務發布成功後，單擊右上方的營運中心。
您也可以單擊左上方的表徵圖，選擇全部產品 > 資料開發與營運 > 營運中心（工作流程）。
單擊左側導覽列中的周期任務營運 > 周期任務，進入周期任務頁面，單擊workshop_start_emr虛節點。
在右側的DAG圖中，按右鍵workshop_start_emr節點，選擇補資料 > 當前節點及下遊節點。
勾選需要補資料的任務，設定業務日期，單擊提交並跳轉。
在補資料頁面單擊重新整理，直至任務全部運行成功即可。

說明

案例完成後，為了避免後續持續產生費用，您可以選擇設定節點調度有效期間或者凍結商務程序根節點（虛擬節點workshop_start_emr）。

後續步驟

資料視覺效果展現：使用者Portrait analysis完成後，使用資料分析模組，將加工後的資料以圖表形式直觀展示，便於您快速提取關鍵資訊，洞察資料背後的業務趨勢。
監控資料品質：為資料加工產生的表配置資料品質監控，提前識別髒資料並進行攔截，避免髒資料影響擴大。
管理資料：使用者Portrait analysis任務流程完成後，在EMR Hive節點內將建立對應資料表。產生的資料表可在資料地圖模組進行查看，可通過血緣查看產生表之間的關係。
API資料服務：擷取最終加工後的資料後，使用資料服務模組，通過標準化的資料服務介面，實現資料的共用與應用，為其他使用API接收資料的業務模組提供資料。