E-MapReduce:データの処理

HiveQL コードの追加

dwd_log_info_di_emr ノードをダブルクリックして構成タブを開き、以下の文を入力します。

-- クリーニング済みログデータ用の DWD レイヤーテーブルを作成します。
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 'サーバーから返された HTTP ステータスコード',
  bytes    STRING COMMENT 'クライアントに返されたバイト数',
  region   STRING COMMENT 'UDF を使用して IP アドレスから導出したリージョン',
  method   STRING COMMENT 'HTTP メソッド (GET、POST など)',
  url      STRING COMMENT 'リクエスト URL パス',
  protocol STRING COMMENT 'HTTP バージョン',
  referer  STRING COMMENT '参照元 URL (ドメインのみ)',
  device   STRING COMMENT 'デバイスタイプ：android、iphone、ipad、macintosh、windows_phone、windows_pc、または unknown',
  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;

-- 現在の業務日付に対する生ログデータの解析およびクリーニングを行います。
-- 各生ログ行は、##@@ で区切られた単一の文字列です。
-- フィールド順序：[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：IP アドレスをリージョン名に変換
  , regexp_substr(request, '(^[^ ]+ )')                     AS method   -- リクエスト文字列から HTTP メソッドを抽出
  , regexp_extract(request, '^[^ ]+ (.*) [^ ]+$')           AS url      -- リクエスト文字列から URL パスを抽出
  , regexp_extract(request, '.* ([^ ]+$)')                   AS protocol -- リクエスト文字列から HTTP バージョンを抽出
  , regexp_extract(referer, '^[^/]+://([^/]+){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   -- User-Agent 文字列から導出
  , 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 -- User-Agent 文字列および URL パスから導出
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;

スケジューリングプロパティの構成

スケジューリング周期 は 日 に設定されています。スケジュール時刻 フィールドはスキップしてください — 実行時刻は、毎日 00:30 以降にすべての子孫ノードをトリガーするルートノード workshop_start_emr から継承されます。

これらの設定を保存すると、dwd_log_info_di_emr ノードは、先祖ノード ods_raw_log_d_emr が OSS からのデータ同期を完了した時点で自動的に実行されます。

ノード構成の保存

上部ツールバーの アイコンをクリックして、ノードを保存します。

HiveQL コードの追加

dws_user_info_all_di_emr ノードをダブルクリックして構成タブを開き、以下の文を入力します。

-- ログ動作とユーザー人口統計情報を統合する DWS レイヤーテーブルを作成します。
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 '参照元ドメイン',
  `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}');

-- クリーニング済みログデータ（左）とユーザー人口統計情報（右）を uid で結合します。
-- COALESCE は、uid がログに存在しユーザー情報テーブルに存在しない場合に対応します。
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;

スケジューリングプロパティの構成

スケジューリング周期 は 日 に設定されています。スケジュール時刻 フィールドはスキップしてください — 実行時刻は、ルートノード workshop_start_emr によって決定されます。

これらの設定を保存すると、dws_user_info_all_di_emr ノードは、両方の先祖ノード（ods_user_info_d_emr および dwd_log_info_di_emr）が完了した時点で自動的に実行されます。

ノード構成の保存

上部ツールバーの アイコンをクリックして、ノードを保存します。

HiveQL コードの追加

ads_user_info_1d_emr ノードをダブルクリックして構成タブを開き、以下の文を入力します。

-- 日次ユーザープロファイル用の ADS レイヤーテーブルを作成します。
CREATE TABLE IF NOT EXISTS ads_user_info_1d_emr (
  uid       STRING COMMENT 'ユーザー ID',
  region    STRING COMMENT 'IP アドレスから導出したリージョン',
  device    STRING COMMENT 'デバイスタイプ',
  pv        BIGINT COMMENT 'ページビュー数',
  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}');

-- 現在の業務日付に対するユーザーごとの動作を集計します。
-- COUNT(0) は、各 uid ごとのログ行数をページビュー総数としてカウントします。
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;

スケジューリングプロパティの構成

スケジューリング周期 は 日 に設定されています。スケジュール時刻 フィールドはスキップしてください — 実行時刻は、ルートノード workshop_start_emr によって決定されます。

これらの設定を保存すると、ads_user_info_1d_emr ノードは、dws_user_info_all_di_emr の完了後に自動的に実行されます。

ノード構成の保存

上部ツールバーの アイコンをクリックして、ノードを保存します。