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 が属するバケットのネットワークアドレスは、[クラシックネットワーク (内部ネットワーク) 経由の Elastic Compute Service (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 'ユーザー行動分析ケース - ユーザー Web サイトアクセスログのファクトテーブル' 
   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. デバッグパラメータを構成します。

   ノードの構成タブの右側ナビゲーション・ペインで、[デバッグ構成] をクリックします。 デバッグ構成タブで、次のパラメータを構成します。 これらのパラメータは、ステップ 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 'ユーザー行動分析ケース - ユーザーの Web サイトアクセス情報のワイドテーブル' 
   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. デバッグパラメータを構成します。

   ノードの構成タブの右側ナビゲーション・ペインで、[デバッグ構成] をクリックします。 デバッグ構成タブで、次のパラメータを構成します。 これらのパラメータは、ステップ 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 'ページビュー',
   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;
   
   -- シナリオ：次の SQL 文は、ユーザー Web サイトアクセスログを保存する 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. デバッグパラメータを構成します。

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

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

5. オプション。 スケジューリングプロパティを構成します。

   このチュートリアルでは、スケジューリングプロパティに関連するパラメータのデフォルト値を保持できます。 ノード構成タブの右側ナビゲーション・ペインで [プロパティ] をクリックできます。 プロパティタブのパラメータの詳細については、「スケジューリングプロパティ」をご参照ください。

   • [スケジューリングパラメータ]：このチュートリアルでは、ワークフローのスケジューリングパラメータが構成されています。 ワークフローの内部ノードのスケジューリングパラメータを個別に構成する必要はありません。 構成されたスケジューリングパラメータは、内部ノードに基づいて開発されたコードとタスクに直接使用できます。

   • [スケジューリングポリシー]：[遅延実行時間] パラメータを構成して、バッチ同期ノードの実行がワークフローの実行よりも遅れる期間を指定できます。 このチュートリアルでは、このパラメータを構成する必要はありません。

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