DataWorks:データ処理

リソースのアップロード（ip2region-starrocks.jar）

1. ip2region-starrocks.jar パッケージをダウンロードします。

   説明

   ip2region-starrocks.jar パッケージは、本チュートリアル専用です。

2. OSS へのリソースのアップロードを行います。

   1. OSS コンソール にログインし、[バケット] ページに移動します。[バケット] ページで、「環境の準備」時に作成したバケットを検索し、バケットのパス内に dataworks_starrocks ディレクトリを作成します。

   2. ip2region-starrocks.jar パッケージを dataworks_starrocks ディレクトリにアップロードします。

      本チュートリアルでは、パッケージの完全な保存パスは https://test.oss-cn-shanghai-internal.aliyuncs.com/dataworks_starrocks/ip2region-starrocks.jar です。アップロード済み OSS リソースの保存パスは、この完全なパスから取得できます。

      説明

      • 本チュートリアルでは、test という名前のバケットを使用します。

      • UDF が属するバケットのネットワークアドレスは、ECS からのクラシックネットワーク（イントラネット）経由のアクセスに使用されるアドレスです。

      • 内部エンドポイントを使用する場合、OSS バケットは DataWorks ワークスペースと同じリージョンに存在している必要があります。本例では中国 (上海) リージョンを使用しています。

関数の登録（getregion）

1. 関数登録用の StarRocks ノードを作成します。

   DataWorks コンソールにログインし、Data Studio ページの DATA STUDIO ウィンドウに移動します。DATA STUDIO ウィンドウの ワークスペースディレクトリ セクションで、 アイコンをクリックし、ノードの作成 > データベース > StarRocks を選択して StarRocks ノードを作成します。

2. 関数登録用のコードを記述します。

   • 関数を登録します。

     CREATE FUNCTION getregion(string)
     RETURNS string
     PROPERTIES ( 
         "symbol" = "com.starrocks.udf.sample.Ip2Region", 
         "type" = "StarrocksJar",
         "file" = "OSS バケットの完全な保存パスを入力します。このパスは前述のサブステップで取得できます。"
     );

   • 関数が正しく登録されたか確認します。

     SELECT getregion('ご利用のローカルマシンの IP アドレス');

   重要

   関数は、開発環境および本番環境それぞれで一度だけ個別に登録できます。本番環境で関数を登録するには、まず StarRocks ノードを本番環境にデプロイする必要があります。

3. StarRocks ノードの構成タブ上部のツールバーで、保存 をクリックします。その後、デプロイ をクリックし、[DEPLOY] タブに表示される指示に従って、StarRocks ノードを開発環境および本番環境の StarRocks 計算リソースにデプロイします。その後、StarRocks ノードのデータをバックフィルして、本番環境における関数登録を完了します。関数登録が完了したら、オペレーションセンターで本番環境の StarRocks ノードを手動で フリーズ します。これにより、関数の重複登録による StarRocks ノードの失敗を防止できます。

dwd_log_info_di_starrocks ノードの構成

本ノードのサンプルコードでは、登録済み関数を用いて先祖テーブル ods_raw_log_d_starrocks のフィールドに対する SQL コードを処理し、テーブルのデータを dwd_log_info_di_starrocks テーブルに同期します。

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

2. ノードの構成タブで、[データソースの選択] ドロップダウンリストから、「環境の準備」時にワークスペースに関連付けた StarRocks 計算リソースを選択します。

3. 以下の SQL 文をコピーし、コードエディターに貼り付けます：

   説明

   dwd_log_info_di_starrocks ノードのサンプルコードでは、登録済み関数を用いて先祖テーブル ods_raw_log_d_starrocks のフィールドに対する SQL コードを処理し、テーブルのデータを dwd_log_info_di_starrocks テーブルに同期します。

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

   CREATE TABLE IF NOT EXISTS dwd_log_info_di_starrocks (
       uid STRING COMMENT 'ユーザー ID',
       ip STRING COMMENT 'IP アドレス',
       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 'ソース URL',
       device STRING COMMENT '端末タイプ',
       identity STRING COMMENT 'アクセスタイプ（クローラー、フィード、ユーザー、不明のいずれか）',
       dt DATE NOT NULL COMMENT '時刻'
   ) DUPLICATE KEY(uid) 
   COMMENT 'ユーザービヘイビア分析事例 - ユーザーのウェブサイトアクセスログのファクトテーブル' 
   PARTITION BY(dt) 
   PROPERTIES ("replication_num" = "1");
   
   -- 本例では、dt フィールドに基づいてフィールドデータが動的にパーティション化されます。ノードの再実行およびデータの重複書き込みを防止するため、以下の SQL 文を用いて、各処理操作の前に既存の宛先パーティションを削除します。
   ALTER TABLE dwd_log_info_di_starrocks DROP PARTITION IF EXISTS p${var} FORCE;
   
   -- シナリオ：以下の SQL 文では、getregion 関数を用いて生ログデータ内の IP アドレスを解析し、正規表現を用いて生データを分析可能なフィールドに分割し、それらを dwd_log_info_di_starrocks テーブルに書き込みます。
   -- 注意：
   --     1. DataWorks ノードで UDF を使用するには、あらかじめ関数を登録する必要があります。
   --     2. スケジューリングシナリオでは、DataWorks のノードに対してスケジューリングパラメーターを構成することで、増分データを所望のテーブルの関連パーティションに毎日同期できます。
   --        実際の開発シナリオでは、ノードのコード内で ${変数名} 形式で変数を定義し、ノードの構成タブの [プロパティ] タブでスケジューリングパラメーターを変数に割り当てることができます。これにより、スケジューリングパラメーターの設定に基づき、ノードコード内のスケジューリングパラメーターの値を動的に置き換えることができます。
   INSERT INTO dwd_log_info_di_starrocks 
   SELECT 
       uid
       , ip  
       , time
       , status
       , bytes 
       , getregion(ip) AS region -- UDF を用いて IP アドレスに基づいてリージョンを取得します。
       ,REGEXP_EXTRACT(request, '([^ ]+)', 1) AS method
       ,REGEXP_EXTRACT(request, '^[^ ]+ (.*) [^ ]+$', 1) AS url
       ,REGEXP_EXTRACT(request, '([^ ]+)$', 1) AS protocol
       ,REGEXP_EXTRACT(referer, '^[^/]+://([^/]+)', 1) AS referer
     , CASE
       WHEN LOWER(agent) REGEXP 'android' THEN 'android'
       WHEN LOWER(agent) REGEXP 'iphone' THEN 'iphone'
       WHEN LOWER(agent) REGEXP 'ipad' THEN 'ipad'
       WHEN LOWER(agent) REGEXP 'macintosh' THEN 'macintosh'
       WHEN LOWER(agent) REGEXP 'windows phone' THEN 'windows_phone'
       WHEN LOWER(agent) REGEXP 'windows' THEN 'windows_pc'
       ELSE 'unknown'
   END AS device
     , CASE
       WHEN LOWER(agent) REGEXP '(bot|spider|crawler|slurp)' THEN 'crawler'
       WHEN LOWER(agent) REGEXP 'feed' OR REGEXP_EXTRACT(request, '^[^ ]+ (.*) [^ ]+$', 0) REGEXP 'feed' THEN 'feed'
       WHEN NOT (LOWER(agent) REGEXP '(bot|spider|crawler|feed|slurp)') 
            AND agent REGEXP '^(Mozilla|Opera)' 
            AND NOT (REGEXP_EXTRACT(request, '^[^ ]+ (.*) [^ ]+$', 0) REGEXP 'feed') THEN 'user'
       ELSE 'unknown'
   END AS identity,
    cast('${var}' AS DATE )AS dt
     FROM (
       SELECT
         SPLIT_PART(CAST(col AS VARCHAR(65533)), '##@@', 1)  AS ip
       , SPLIT_PART(CAST(col AS VARCHAR(65533)), '##@@', 2)  AS uid
       , SPLIT_PART(CAST(col AS VARCHAR(65533)), '##@@', 3)  AS time
       , SPLIT_PART(CAST(col AS VARCHAR(65533)), '##@@', 4)  AS request
       , SPLIT_PART(CAST(col AS VARCHAR(65533)), '##@@', 5)  AS status
       , SPLIT_PART(CAST(col AS VARCHAR(65533)), '##@@', 6)  AS bytes
       , SPLIT_PART(CAST(col AS VARCHAR(65533)), '##@@', 7)  AS referer
       , SPLIT_PART(CAST(col AS VARCHAR(65533)), '##@@', 8)  AS agent
   FROM
       ods_raw_log_d_starrocks
   WHERE
       dt = '${var}'
   ) a;

4. デバッグパラメーターを構成します。

   ノードの構成タブ右側のナビゲーションペインで、Run Configuration をクリックします。タブで以下のパラメーターを構成します。これらのパラメーターは、ステップ 4 でワークフローをテストするために使用します。

   パラメーター	説明
   計算リソース	「環境の準備」時にワークスペースに関連付けた StarRocks 計算リソースを選択します。
   リソースグループ	「環境の準備」時に購入したサーバーレスリソースグループを選択します。
   スクリプトパラメーター	[var] パラメーターの [パラメーター値] 列に、yyyymmdd 形式の定数値を入力します。例：var=20250223。ワークフローのデバッグ時には、Data Studio がワークフロー内のノードで定義された変数をこの定数で置き換えます。

5. （任意）スケジューリングプロパティを構成します。

   本チュートリアルでは、スケジューリングパラメーターをデフォルト値のままにしておくことができます。これらを表示するには、ノードエディター右側のペインで スケジューリング構成 をクリックします。各パラメーターの詳細については、「ノードのスケジューリング構成」をご参照ください。

   • スケジューリングパラメーター：これらのパラメーターはワークフローから継承され、個別のノードに対して構成する必要はありません。タスクやコードで直接使用できます。

   • [スケジューリングポリシー]: [遅延実行時間] パラメーターを使用して、ワークフロー の開始後からサブノード を実行するまでの待機時間を指定できます。このパラメーターは、本チュートリアルでは設定されません。

6. 構成タブ上部のツールバーで、保存 をクリックしてノードを保存します。

dws_user_info_all_di_starrocks ノードの構成

このノードは、基本ユーザー情報テーブル ods_user_info_d_starrocks およびログデータテーブル dwd_log_info_di_starrocks のデータを集約し、集約結果を dws_user_info_all_di_starrocks テーブルに同期するために使用します。

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

2. ノードの構成タブで、[データソースの選択] ドロップダウンリストから、「環境の準備」時にワークスペースに関連付けた StarRocks 計算リソースを選択します。

3. 以下の SQL 文をコピーし、コードエディターに貼り付けます：

   説明

   dws_user_info_all_di_starrocks ノードの構成タブで、先祖テーブル dwd_log_info_di_starrocks および ods_user_info_d_starrocks のデータを集約し、集約結果を dws_user_info_all_di_starrocks テーブルに同期するコードを記述します。

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

   CREATE TABLE IF NOT EXISTS dws_user_info_all_di_starrocks (
       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 'ソース URL',
       TIME STRING COMMENT 'yyyymmddhh:mi:ss 形式の時刻',
       dt DATE NOT NULL COMMENT '時刻'
   ) DUPLICATE KEY(uid) 
   COMMENT 'ユーザービヘイビア分析事例 - ユーザーのウェブサイトアクセス情報を統合したワイドテーブル' 
   PARTITION BY(dt) 
   PROPERTIES ("replication_num" = "1");
   
   -- 本例では、dt フィールドに基づいてフィールドデータが動的にパーティション化されます。ノードの再実行およびデータの重複書き込みを防止するため、以下の SQL 文を用いて、各処理操作の前に既存の宛先パーティションを削除します。
   ALTER TABLE dws_user_info_all_di_starrocks DROP PARTITION IF EXISTS p${var} FORCE;
   
   
   -- シナリオ：dwd_log_info_di_starrocks テーブルに格納された処理済みログデータと ods_user_info_d_starrocks テーブルに格納された基本ユーザー情報をマージし、マージ済みデータを dws_user_info_all_di_starrocks テーブルに同期します。
   -- 注意：スケジューリングシナリオでは、DataWorks のノードに対してスケジューリングパラメーターを構成することで、増分データを所望のテーブルの関連パーティションに毎日同期できます。
   -- 実際の開発シナリオでは、ノードのコード内で ${変数名} 形式で変数を定義し、ノードの構成タブの [プロパティ] タブでスケジューリングパラメーターを変数に割り当てることができます。これにより、スケジューリングパラメーターの設定に基づき、ノードコード内のスケジューリングパラメーターの値を動的に置き換えることができます。
   INSERT INTO dws_user_info_all_di_starrocks 
   SELECT 
       IFNULL(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,
       a.dt
   FROM dwd_log_info_di_starrocks a
   LEFT JOIN ods_user_info_d_starrocks b
   ON a.uid = b.uid
   WHERE a.dt = '${var}';
   

4. デバッグパラメーターを構成します。

   ノードの構成タブ右側のナビゲーションペインで、Run Configuration をクリックします。タブで以下のパラメーターを構成します。これらのパラメーターは、ステップ 4 でワークフローをテストするために使用します。

   パラメーター	説明
   計算リソース	「環境の準備」時にワークスペースに関連付けた StarRocks 計算リソースを選択します。
   リソースグループ	「環境の準備」時に購入したサーバーレスリソースグループを選択します。
   スクリプトパラメーター	[var] パラメーターの [パラメーター値] 列に、yyyymmdd 形式の定数値を入力します。例：var=20250223。ワークフローのデバッグ時には、Data Studio がワークフロー内のノードで定義された変数をこの定数で置き換えます。

5. （任意）スケジューリングプロパティを構成します。

   本チュートリアルでは、スケジューリングパラメーターをデフォルト値のままにしておくことができます。これらを表示するには、ノードエディター右側のペインで スケジューリング構成 をクリックします。各パラメーターの詳細については、「ノードのスケジューリング構成」をご参照ください。

   • スケジューリングパラメーター：これらのパラメーターはワークフローから継承され、個別のノードに対して構成する必要はありません。タスクやコードで直接使用できます。

   • [スケジューリングポリシー]: [遅延実行時間] パラメーターを使用すると、ワークフローの開始後、サブノードを実行するまでの待機時間を指定できます。このパラメーターは、本チュートリアルでは設定されていません。

6. 構成タブ上部のツールバーで、保存 をクリックしてノードを保存します。

ads_user_info_1d_starrocks ノードの構成

このノードは、dws_user_info_all_di_starrocks テーブルのデータをさらに処理し、処理済みデータを ads_user_info_1d_starrocks テーブルに同期して基本的なユーザープロファイルを生成するために使用します。

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

2. ノードの構成タブで、[データソースの選択] ドロップダウンリストから、「環境の準備」時にワークスペースに関連付けた StarRocks 計算リソースを選択します。

3. 以下の SQL 文をコピーし、コードエディターに貼り付けます：

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

   CREATE TABLE IF NOT EXISTS ads_user_info_1d_starrocks (
   uid STRING COMMENT 'ユーザー ID',
   region STRING COMMENT 'IP アドレスに基づいて取得されたリージョン',
   device STRING COMMENT '端末タイプ',
   pv BIGINT COMMENT 'PV',
   gender STRING COMMENT '性別',
   age_range STRING COMMENT '年齢層',
   zodiac STRING COMMENT '星座',
   dt DATE NOT NULL COMMENT '時刻'
   ) DUPLICATE KEY(uid) 
   COMMENT 'ユーザービヘイビア分析事例 - ユーザープロファイルデータ' 
   PARTITION BY(dt) 
   PROPERTIES ("replication_num" = "1");
   
   -- 本例では、dt フィールドに基づいてフィールドデータが動的にパーティション化されます。ノードの再実行およびデータの重複書き込みを防止するため、以下の SQL 文を用いて、各処理操作の前に既存の宛先パーティションを削除します。
   ALTER TABLE ads_user_info_1d_starrocks DROP PARTITION IF EXISTS p${var} FORCE;
   
   -- シナリオ：ユーザーのウェブサイトアクセスログを格納する dws_user_info_d_all_di_starrocks ワイドテーブルのデータをさらに処理し、基本的なユーザープロファイルデータを生成し、ads_user_info_1d_starrocks テーブルに同期します。
   -- 注意：スケジューリングシナリオでは、DataWorks のノードに対してスケジューリングパラメーターを構成することで、増分データを所望のテーブルの関連パーティションに毎日同期できます。
   -- 実際の開発シナリオでは、ノードのコード内で ${変数名} 形式で変数を定義し、ノードの構成タブの [プロパティ] タブでスケジューリングパラメーターを変数に割り当てることができます。これにより、スケジューリングパラメーターの設定に基づき、ノードコード内のスケジューリングパラメーターの値を動的に置き換えることができます。
   INSERT INTO ads_user_info_1d_starrocks 
   SELECT 
   uid,
   MAX(region) AS region,
   MAX(device) AS device,
   COUNT(*) AS pv,
   MAX(gender) AS gender,
   MAX(age_range) AS age_range,
   MAX(zodiac) AS zodiac,
   dt
   FROM dws_user_info_all_di_starrocks
   WHERE dt = '${var}'
   GROUP BY uid, dt;
   
   SELECT * FROM dws_user_info_all_di_starrocks
   WHERE dt = '${var}';

4. デバッグパラメーターを構成します。

   ノードの構成タブ右側のナビゲーションペインで、Run Configuration をクリックします。タブで以下のパラメーターを構成します。これらのパラメーターは、ステップ 4 でワークフローをテストするために使用します。

   パラメーター	説明
   計算リソース	「環境の準備」時にワークスペースに関連付けた StarRocks 計算リソースを選択します。
   リソースグループ	「環境の準備」時に購入したサーバーレスリソースグループを選択します。
   スクリプトパラメーター	[var] パラメーターの [パラメーター値] 列に、yyyymmdd 形式の定数値を入力します。例：var=20250223。ワークフローのデバッグ時には、Data Studio がワークフロー内のノードで定義された変数をこの定数で置き換えます。

5. （任意）スケジューリングプロパティを構成します。

   本チュートリアルでは、スケジューリングパラメーターをデフォルト値のままにしておくことができます。これらを表示するには、ノードエディター右側のペインで スケジューリング構成 をクリックします。各パラメーターの詳細については、「ノードのスケジューリング構成」をご参照ください。

   • スケジューリングパラメーター：これらのパラメーターはワークフローから継承され、個別のノードに対して構成する必要はありません。タスクやコードで直接使用できます。

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

6. 構成タブ上部のツールバーで、保存 をクリックしてノードを保存します。