E-MapReduce:データの処理

ノードコードの構成

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;

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

構成の保存

この例では、ビジネス要件に基づいて他の必要な設定項目を構成できます。構成が完了したら、ノードの構成タブの上部ツールバーにある アイコンをクリックして、ノードの構成を保存します。

ログテーブルの分割結果の検証

先祖ノードと現在のノードが正常に実行された後、DataStudio ページの左側のナビゲーションウィンドウで [アドホッククエリ] をクリックします。アドホッククエリウィンドウで、[EMR Spark SQL] タイプのアドホッククエリタスクを作成し、SQL ステートメントを記述して、現在のノードによって作成されたテーブルが期待どおりに生成されたかどうかを確認します。

-- パーティションフィルター条件を現在の操作のデータタイムスタンプに更新する必要があります。たとえば、ノードが 2023 年 2 月 22 日に実行されるようにスケジュールされている場合、ノードのデータタイムスタンプは 20230221 であり、ノードのスケジューリング時間より 1 日早くなります。
SELECT * FROM dwd_log_info_di_spark WHERE dt ='データタイムスタンプ';

ノードコードの構成

dws_user_info_all_di_spark ノードをダブルクリックして、ノードの構成タブに移動します。構成タブで、次のステートメントを記述します。

-- シナリオ: 以下のサンプルコードの SQL ステートメントは Spark SQL ステートメントです。dwd_log_info_di_spark テーブルと ods_user_info_d_spark テーブルを uid フィールドに基づいて結合し、指定された 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}');

-- 基本的なユーザー情報テーブルと新しいログテーブルからデータを挿入
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 
   dwd_log_info_di_spark as a
LEFT OUTER JOIN 
    ods_user_info_d_spark as b
ON 
    a.uid = b.uid ;

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

構成の保存

この例では、ビジネス要件に基づいて他の必要な設定項目を構成できます。構成が完了したら、ノードの構成タブの上部ツールバーにある アイコンをクリックして、ノードの構成を保存します。

データマージ結果の検証

先祖ノードと現在のノードが正常に実行された後、DataStudio ページの左側のナビゲーションウィンドウで [アドホッククエリ] をクリックします。アドホッククエリウィンドウで、[EMR Spark SQL] タイプのアドホッククエリタスクを作成し、SQL ステートメントを記述して、現在のノードによって作成されたテーブルが期待どおりに生成されたかどうかを確認します。

-- パーティションフィルター条件を現在の操作のデータタイムスタンプに更新する必要があります。たとえば、ノードが 2024 年 8 月 8 日に実行されるようにスケジュールされている場合、ノードのデータタイムスタンプは 20240807 であり、ノードのスケジューリング時間より 1 日早くなります。
SELECT * FROM dws_user_info_all_di_spark WHERE dt ='データタイムスタンプ';

ノードコードの構成

ads_user_info_1d_spark ノードをダブルクリックして、ノードの構成タブに移動します。構成タブで、次のステートメントを記述します。

-- シナリオ: 以下のサンプルコードの SQL ステートメントは Spark SQL ステートメントです。Spark SQL 関数を使用して、Spark SQL で 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;

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

構成の保存

この例では、ビジネス要件に基づいて他の必要な設定項目を構成できます。構成が完了したら、ノードの構成タブの上部ツールバーにある アイコンをクリックして、ノードの構成を保存します。

ユーザープロファイルテーブルの結果の検証

先祖ノードと現在のノードが正常に実行された後、DataStudio ページの左側のナビゲーションウィンドウで [アドホッククエリ] をクリックします。アドホッククエリウィンドウで、[EMR Spark SQL] タイプのアドホッククエリタスクを作成し、SQL ステートメントを記述して、現在のノードによって作成されたテーブルが期待どおりに生成されたかどうかを確認します。

-- パーティションフィルター条件を現在の操作のデータタイムスタンプに更新する必要があります。たとえば、ノードが 2023 年 2 月 22 日に実行されるようにスケジュールされている場合、ノードのデータタイムスタンプは 20230221 であり、ノードのスケジューリング時間より 1 日早くなります。
SELECT * FROM ads_user_info_1d_spark WHERE dt ='データタイムスタンプ';