DataWorks:データの処理

ワークフローのキャンバスで、dwd_log_info_di_spark ノードにポインターを移動し、[ノードを開く] をクリックします。

次の SQL 文をコピーして、コードエディタに貼り付けます:

-- シナリオ: 次のサンプルコードの 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)
LOCATION 'oss://dw-spark-demo.oss-cn-shanghai-internal.aliyuncs.com/dwd_log_info_di_spark/log_${bizdate}/';

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;

ノードの設定タブの右側のナビゲーションウィンドウで、[デバッグ設定] をクリックします。[デバッグ設定] タブで、次のパラメーターを設定します。これらのパラメーターは、ステップ 4 でワークフローをテストするために使用されます。

(オプション) スケジューリングプロパティを設定します。

このチュートリアルでは、スケジューリングパラメーターのデフォルト値を維持できます。ノード編集ページの右側のペインで [スケジューリング設定] をクリックできます。パラメーターの詳細については、「ノードのスケジューリングプロパティを設定する」をご参照ください。

• スケジューリングパラメーター: これらは、このチュートリアルのワークフローですでに設定されています。内部ノード用に設定する必要はありません。タスクやコードで直接使用できます。

• スケジューリングポリシー: 子ノードがワークフローの開始後に実行を待機する時間を指定するには、[遅延実行時間] パラメーターを設定できます。このチュートリアルでは、このパラメーターを設定する必要はありません。

[保存] をクリックします。

ワークフローのキャンバスで、dws_user_info_all_di_spark ノードにポインターを移動し、[ノードを開く] をクリックします。

次の SQL 文をコピーして、コードエディタに貼り付けます:

-- シナリオ: 次のサンプルコードの 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)
LOCATION 'oss://dw-spark-demo.oss-cn-shanghai-internal.aliyuncs.com/dws_user_info_all_di_spark/log_${bizdate}/';

-- パーティションを追加します。
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 (
  SELECT * 
  FROM dwd_log_info_di_spark 
  WHERE dt='${bizdate}'
) a
LEFT OUTER JOIN (
  SELECT * 
  FROM ods_user_info_d_spark 
  WHERE dt='${bizdate}'
) b
ON 
    a.uid = b.uid;

ノードの設定タブの右側のナビゲーションウィンドウで、[デバッグ設定] をクリックします。[デバッグ設定] タブで、次のパラメーターを設定します。これらのパラメーターは、ステップ 4 でワークフローをテストするために使用されます。

(オプション) スケジューリングプロパティを設定します。

このチュートリアルでは、スケジューリングパラメーターのデフォルト値を維持できます。ノード編集ページの右側のペインで [スケジューリング設定] をクリックできます。パラメーターの詳細については、「ノードのスケジューリングプロパティを設定する」をご参照ください。

• スケジューリングパラメーター: これらは、このチュートリアルのワークフローですでに設定されています。内部ノード用に設定する必要はありません。タスクやコードで直接使用できます。

• スケジューリングポリシー: 子ノードがワークフローの開始後に実行を待機する時間を指定するには、[遅延実行時間] パラメーターを設定できます。このチュートリアルでは、このパラメーターを設定する必要はありません。

[保存] をクリックします。

ワークフローのキャンバスで、ads_user_info_1d_spark ノードにポインターを移動し、[ノードを開く] をクリックします。

次の SQL 文をコピーして、コードエディタに貼り付けます:

-- シナリオ: 次のサンプルコードの 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
)
LOCATION 'oss://dw-spark-demo.oss-cn-shanghai-internal.aliyuncs.com/ads_user_info_1d_spark/log_${bizdate}/';

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; 

ノードの設定タブの右側のナビゲーションウィンドウで、[デバッグ設定] をクリックします。[デバッグ設定] タブで、次のパラメーターを設定します。これらのパラメーターは、ステップ 4 でワークフローをテストするために使用されます。

(オプション) スケジューリングプロパティを設定します。

このチュートリアルでは、スケジューリングパラメーターのデフォルト値を維持できます。ノード編集ページの右側のペインで [スケジューリング設定] をクリックできます。パラメーターの詳細については、「ノードのスケジューリングプロパティを設定する」をご参照ください。

• スケジューリングパラメーター: これらは、このチュートリアルのワークフローですでに設定されています。内部ノード用に設定する必要はありません。タスクやコードで直接使用できます。

• スケジューリングポリシー: 子ノードがワークフローの開始後に実行を待機する時間を指定するには、[遅延実行時間] パラメーターを設定できます。このチュートリアルでは、このパラメーターを設定する必要はありません。

[保存] をクリックします。