DataWorks:データの処理

ip2region.jar のアップロード

1. ip2region.jar ファイルをダウンロードします。

   説明

   サンプル ip2region.jar リソースはチュートリアル専用です。

2. DataStudio ページの左側のナビゲーションウィンドウで、 アイコンをクリックして リソース管理 ページに移動します。 > New Resource > MaxCompute Jar をクリックし、リソースに名前を付けてアップロードページに進みます。

   説明

   リソース名は、アップロードするファイル名と一致する必要はありません。

3. Document Source で Local を選択します。ファイルコンテンツの横にある Click Upload をクリックし、ダウンロードした ip2region.jar ファイルを選択します。

4. Data Source には、環境の準備 ステップでバインドした MaxCompute コンピューティングリソースを選択します。

5. ツールバーで、Save をクリックし、次に Publish をクリックします。プロンプトに従って、開発環境と本番環境の両方の MaxCompute プロジェクトにリソースを公開します。

getregion の登録

1. リソース管理 ページで、 > New Function > MaxCompute 関数 をクリックし、関数に getregion という名前を付けて登録ページに移動します。

2. [関数登録] ページで、必須パラメーターを設定します。次の表に、このチュートリアルの主要なパラメーターを示します。他のパラメーターはデフォルト値のままにします。

   パラメーター	説明
   Function type	OTHER を選択します。
   Data Source	「環境の準備」ステップでバインドした MaxCompute コンピューティングリソースを選択します。
   Class Name	org.alidata.odps.udf.Ip2Region と入力します。
   Resource List	ip2region.jar を選択します。
   Description	IP アドレスをリージョンに変換します。
   Command Format	getregion('ip') と入力します。
   Parameter Description	IP アドレス。

3. ツールバーで、Saveをクリックし、次にPublishをクリックします。プロンプトに従って、開発環境と本番環境の両方の MaxCompute プロジェクトに関数を公開します。

dwd_log_info_di ノードの設定

このノードのサンプルコードでは、登録した UDF を使用して上流テーブル ods_raw_log_d のフィールドを処理し、その結果を dwd_log_info_di テーブルに書き込みます。

1. Data Studio の左側のナビゲーションウィンドウで、 アイコンをクリックしてデータ開発ページを開きます。Project Directory セクションで、作成したワークフローを見つけ、クリックしてワークフローキャンバスを開きます。

2. ワークフローキャンバスで、dwd_log_info_di ノードにマウスカーソルを合わせ、Open Node をクリックします。

3. 次のコードをノードエディターに貼り付けます。

   dwd_log_info_di ノードのサンプルコード

   -- dwd_log_info_di テーブルを作成します。
   CREATE TABLE IF NOT EXISTS dwd_log_info_di (
    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 'Referer URL',
    device STRING COMMENT 'デバイスタイプ',
    identity STRING COMMENT 'アクセスタイプ：クローラー、フィード、ユーザー、または不明'
   )
   PARTITIONED BY (
    dt STRING
   )
   LIFECYCLE 14;
   
   -- データを処理します。
   -- シナリオ：次の SQL ステートメントは、getregion 関数を使用して生ログデータから IP アドレスを解析します。
   -- また、正規表現を使用して生データを分析可能なフィールドに分割し、dwd_log_info_di テーブルに書き込みます。
   -- getregion 関数はこのチュートリアルで提供されます。
   -- 注：
   -- 1. DataWorks ノードで UDF を使用する前に、必要なリソースをアップロードし、そのリソースを使用して関数を登録する必要があります。
   --    このチュートリアルの getregion 関数のリソースは ip2region.jar です。
   -- 2. DataWorks は、日次増分データをターゲットテーブルの対応するパーティションにロードできるスケジューリングパラメーターを提供します。
   --    開発では、${variable_name} 形式を使用して変数を定義できます。スケジューリングプロパティでは、これらの変数にスケジューリングパラメーターを割り当てて、スケジューリングされた実行中に動的に値を渡すことができます。
   INSERT OVERWRITE TABLE dwd_log_info_di PARTITION (dt='${bizdate}')
   SELECT ip 
     , uid
     , time
     , status
     , bytes 
     , getregion(ip) AS region -- UDF を使用して IP アドレスからリージョンを取得します。
     , regexp_substr(request, '(^[^ ]+ )') AS method -- 正規表現を使用してリクエストを 3 つのフィールドに分割します。
     , regexp_extract(request, '^[^ ]+ (.*) [^ ]+$') AS url
     , regexp_substr(request, '([^ ]+$)') AS protocol 
     , regexp_extract(referer, '^[^/]+://([^/]+){1}') AS referer -- 正規表現を使用して Referer をクリーンアップし、より正確な URL を取得します。
     , CASE
       WHEN TOLOWER(agent) RLIKE 'android' THEN 'android' -- ユーザーエージェントからデバイスとアクセスタイプを取得します。
       WHEN TOLOWER(agent) RLIKE 'iphone' THEN 'iphone'
       WHEN TOLOWER(agent) RLIKE 'ipad' THEN 'ipad'
       WHEN TOLOWER(agent) RLIKE 'macintosh' THEN 'macintosh'
       WHEN TOLOWER(agent) RLIKE 'windows phone' THEN 'windows_phone'
       WHEN TOLOWER(agent) RLIKE 'windows' THEN 'windows_pc'
       ELSE 'unknown'
     END AS device
     , CASE
       WHEN TOLOWER(agent) RLIKE '(bot|spider|crawler|slurp)' THEN 'crawler'
       WHEN TOLOWER(agent) RLIKE 'feed'
       OR regexp_extract(request, '^[^ ]+ (.*) [^ ]+$') RLIKE 'feed' THEN 'feed'
       WHEN TOLOWER(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 time
       , 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  
     WHERE dt ='${bizdate}'
   ) a;

4. 実行時パラメーターの設定

   MaxCompute SQL ノードエディターの右側のペインで、Run Configuration をクリックします。手順 4 のテスト実行用に、次のパラメーターを設定します。

   パラメーター	説明
   Computing Resources	「環境の準備」ステップでバインドした MaxCompute コンピューティングリソースと、それに対応するコンピューティングクォータを選択します。
   Resource Group	「環境の準備」ステップで購入したサーバーレスリソースグループを選択します。
   Script Parameters	このパラメーターは空白のままにします。 サンプルコードでは、${bizdate} を使用してビジネス日付を表します。 ステップ 4 でワークフローを実行する際に、This operation value を 20250223 などの定数に設定すると、タスクが変数をこの定数に置き換えます。

5. (任意) スケジューリングプロパティの設定

   このチュートリアルでは、デフォルトのスケジューリングプロパティをそのまま使用できます。それらを表示するには、MaxCompute SQL ページの右側のペインでScheduling Configurationをクリックします。これらのパラメーターの詳細については、「ノードのスケジューリングプロパティを設定する」をご参照ください。

   • Scheduling Parameters: このチュートリアル用にワークフローレベルで設定済みのパラメーターです。ワークフロー内のノードは、これらのパラメーターをタスクやコードで直接使用できます。

   • Scheduling Policy: Delayed execution time パラメーターを使用して、ワークフローの開始後にノードが実行されるまでの遅延を指定できます。このチュートリアルでは不要です。

6. ツールバーの Save をクリックします。

dws_user_info_all_di ノードの設定

このノードは、ユーザー情報テーブル (ods_user_info_d) と処理済みログテーブル (dwd_log_info_di) のデータを集計し、その結果を dws_user_info_all_di テーブルに書き込みます。

1. ワークフローキャンバスで、dws_user_info_all_di ノードにマウスカーソルを合わせ、Open Nodeをクリックします。

2. 次のコードをノードエディターに貼り付けます。

   dws_user_info_all_di ノードのサンプルコード

   -- dws_user_info_all_di テーブルを作成します。
   CREATE TABLE IF NOT EXISTS dws_user_info_all_di (
    uid STRING COMMENT 'ユーザー ID',
    gender STRING COMMENT '性別',
    age_range STRING COMMENT '年齢層',
    zodiac STRING COMMENT '星座',
    region STRING COMMENT 'リージョン (IP アドレスから派生)',
    device STRING COMMENT 'デバイスタイプ',
    identity STRING COMMENT 'アクセスタイプ：クローラー、フィード、ユーザー、または不明',
    method STRING COMMENT 'HTTP リクエストメソッド',
    url STRING COMMENT 'URL',
    referer STRING COMMENT 'Referer URL',
    time STRING COMMENT '時刻 (yyyymmddhh:mi:ss 形式)'
   )
   PARTITIONED BY (
    dt STRING
   )
   LIFECYCLE 14;
   
   -- データを処理します。
   -- シナリオ：dwd_log_info_di の処理済みログデータと ods_user_info_d のユーザー情報を集計し、
   -- その結果を dws_user_info_all_di テーブルに書き込みます。
   -- 注：DataWorks は、日次増分データをターゲットテーブルの対応するパーティションにロードできるスケジューリングパラメーターを提供します。
   -- 開発では、${variable_name} 形式を使用して変数を定義できます。スケジューリングプロパティでは、これらの変数にスケジューリングパラメーターを割り当てて、スケジューリングされた実行中に動的に値を渡すことができます。
   INSERT OVERWRITE TABLE dws_user_info_all_di  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  
     WHERE dt = '${bizdate}'
   ) a
   LEFT OUTER JOIN (
     SELECT *
     FROM ods_user_info_d
     WHERE dt = '${bizdate}'
   ) b
   ON a.uid = b.uid;

3. 実行時パラメーターの設定

   MaxCompute SQL ノードエディターの右側のペインで、Run Configuration をクリックします。ステップ 4 でのテスト実行用に、以下のパラメーターを設定します。

   パラメーター	説明
   Computing Resources	「環境の準備」ステップでバインドした MaxCompute コンピューティングリソースと、それに対応するコンピューティングクォータを選択します。
   Resource Group	「環境の準備」ステップで購入したサーバーレスリソースグループを選択します。
   Script Parameters	このパラメーターは空白のままにします。サンプルコードでは、${bizdate} を使用してビジネス日付を表します。ステップ 4 でワークフローを実行するときは、This operation value を 20250223 などの定数に設定します。その後、タスクによって変数がこの定数に置き換えられます。

4. (任意) スケジューリングプロパティの設定

   このチュートリアルでは、デフォルトのスケジューリングプロパティをそのまま使用できます。これらを表示するには、MaxCompute SQL ページの右側のペインでScheduling Configurationをクリックします。これらのパラメーターの詳細については、「ノードのスケジューリングプロパティを設定する」をご参照ください。

   • Scheduling Parameters: このチュートリアルでは、ワークフローレベルで既に設定されており、ワークフロー内のノードはタスクとコードでこれらのパラメーターを直接使用できます。

   • Scheduling Policy: Delayed execution time パラメーターを使用して、ワークフロー開始後のノード実行を遅延させることができます。このチュートリアルでは、この設定は不要です。

5. ツールバーで、Save をクリックします。

ads_user_info_1d ノードの設定

このノードは、dws_user_info_all_di テーブルのデータをさらに処理し、最終結果を基本ユーザープロファイルを含む ads_user_info_1d テーブルに書き込みます。

1. ワークフローキャンバスで、ads_user_info_1d ノードにカーソルを合わせ、Open Node をクリックします。

2. 次のコードをノードエディターに貼り付けます。

   ads_user_info_1d ノードのサンプルコード

   -- ads_user_info_1d テーブルを作成します。
   CREATE TABLE IF NOT EXISTS ads_user_info_1d (
    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
   )
   LIFECYCLE 14;    
   
   -- データを処理します。
   -- シナリオ：ユーザーアクセス情報ワイドテーブル (dws_user_info_all_di) をさらに処理して、
   -- 基本ユーザープロファイルデータを生成し、ads_user_info_1d テーブルに書き込みます。
   -- 注：DataWorks は、日次増分データをターゲットテーブルの対応するパーティションにロードできるスケジューリングパラメーターを提供します。
   -- 開発では、${variable_name} 形式を使用して変数を定義できます。スケジューリングプロパティでは、これらの変数にスケジューリングパラメーターを割り当てて、スケジューリングされた実行中に動的に値を渡すことができます。
   INSERT OVERWRITE TABLE ads_user_info_1d  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
   WHERE dt = '${bizdate}'
   GROUP BY uid; 

3. 実行時パラメーターの設定

   MaxCompute SQL ノードエディターの右側のペインで、Run Configuration をクリックして、ステップ 4 のテスト実行用に次のパラメーターを設定します。

   パラメーター	説明
   Computing Resources	「環境の準備」ステップでバインドした MaxCompute コンピューティングリソースと、それに対応するコンピューティングクォータを選択します。
   Resource Group	「環境の準備」ステップで購入したサーバーレスリソースグループを選択します。
   Script Parameters	このパラメーターは空白のままにします。 サンプルコードでは、${bizdate} を使用してビジネス日付を表します。 ステップ 4 でワークフローを実行するときは、This operation value を 20250223 などの定数に設定します。 そうすると、タスクによって変数がこの定数に置き換えられます。

4. (任意) スケジューリングプロパティの設定

   このチュートリアルでは、デフォルトのスケジューリングプロパティをそのまま使用できます。これらを表示するには、MaxCompute SQL ページの右側のペインでScheduling Configurationをクリックします。これらのパラメーターの詳細については、「ノードのスケジューリングプロパティを設定する」をご参照ください。

   • Scheduling Parameters：これらは、このチュートリアルではワークフローレベルで設定済みです。ワークフロー内のノードは、タスクとコードでこれらのパラメーターを直接使用できます。

   • Scheduling Policy: Delayed execution time パラメーターを使用して、ワークフローが開始してからノードが実行されるまでの遅延を指定できます。このチュートリアルでは、これは必須ではありません。

5. ツールバーで、Save をクリックします。