DataWorks:データ処理

ノードコードを構成します。

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 ステートメントです。 uid フィールドに基づいて dwd_log_info_di_spark テーブルと ods_user_info_d_spark テーブルを結合し、指定された 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 'ページビュー',
  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 ='データタイムスタンプ';