MaxCompute SQL ノードによるユーザープロファイルパイプラインの作成 - DataWorks

本記事では、ods_user_info_d（ユーザー情報）および ods_raw_log_d（アクセスログ）の各テーブルを MaxCompute に同期した後の処理方法について説明します。DataWorks の MaxCompute ノードを使用して、ユーザープロファイルデータを生成します。本ガイドでは、シンプルなデータウェアハウス構成における DataWorks および MaxCompute を用いたデータ計算・分析の手順を示します。

前提条件

開始する前に、「データの同期」で説明されている手順を完了してください。

データ処理パイプラインの構築

前段階では、データを MaxCompute に同期しました。本段階では、このデータを処理して基本的なユーザープロファイルデータを生成します。

DataStudio の左側ナビゲーションウィンドウでをクリックします。プロジェクトディレクトリ から作成済みのワークフローを見つけ、クリックしてワークフローキャンバスを開きます。

以下の表は、本チュートリアルで使用されるサンプルノードとその役割を示しています。

ノードタイプ	ノード名	ノードの役割
MaxCompute SQL	`dwd_log_info_di`	ビルトイン関数およびユーザー定義関数 (UDF) (`getregion`) を使用して、`ods_raw_log_d` テーブルから生ログデータを解析し、`dwd_log_info_di` テーブルの複数のフィールドに出力を書き込みます。
MaxCompute SQL	`dws_user_info_all_di`	ユーザー情報テーブル (`ods_user_info_d`) および処理済みログデータテーブル (`dwd_log_info_di`) からデータを集約し、その結果を `dws_user_info_all_di` テーブルに書き込みます。
MaxCompute SQL	`ads_user_info_1d`	`dws_user_info_all_di` テーブルからデータを処理し、基本的なユーザープロファイルを `ads_user_info_1d` テーブルに書き込みます。

ノードをドラッグ＆ドロップして接続し、依存関係を設定します。最終的なワークフローは次のようになります：
説明
ワークフロー内では、子ノードのコードを解析することでシステムが自動的に依存関係を特定することもできますが、本チュートリアルではノード間の依存関係を手動で接続する方法を採用します。詳細については、「上流および下流の依存関係」および「自動解析メカニズム」をご参照ください。

ユーザー定義関数の登録

同期済みログデータを構造化されたテーブル形式に解析するために、getregion ユーザー定義関数 (UDF) を登録します。

重要

本チュートリアルでは、IP アドレスを地理的リージョンに変換する関数に必要なリソースを提供します。このリソースをダウンロードし、DataWorks ワークスペースにアップロードした後、関数を登録してください。
この関数はチュートリアル専用であり、サンプル IP リソースを使用します。本番環境で IP アドレスを地理的位置にマッピングする必要がある場合は、専門の IP データプロバイダーから IP 変換サービスを取得してください。

リソース (ip2region.jar) のアップロード

ip2region.jar をダウンロードします。
説明
ip2region.jar リソースは、チュートリアル専用のサンプルです。
DataStudio ページの左側ナビゲーションウィンドウでをクリックします。 > リソースの作成 > MaxCompute > Jar をクリックし、リソース名を設定します。
説明
リソース名は、アップロードするファイル名と一致させる必要はありません。
ソース で ローカル を選択します。アップロードをクリック をクリックし、ダウンロードした ip2region.jar ファイルを選択します。
データソース で、「環境の準備」の前段階でバインドした MaxCompute 計算リソースを選択します。
ツールバーで保存をクリックし、その後公開をクリックします。指示に従って、リソースを開発環境および本番環境の MaxCompute プロジェクトに公開します。

関数 (getregion) の登録

Resource Management ページで、 > 関数の作成 > MaxCompute > 関数 をクリックし、関数名を getregion に設定します。

関数の登録 ページでパラメーターを構成します。以下の表は、本チュートリアルで重要なパラメーターを示しています。その他のパラメーターはデフォルト値のままとします。

パラメーター	説明
関数タイプ	`OTHER` を選択します。
データソース	「環境の準備」の前段階でバインドした MaxCompute 計算リソースを選択します。
クラス名	`org.alidata.odps.udf.Ip2Region` を入力します。
リソース一覧	`ip2region.jar` を選択します。
説明	IP アドレスをリージョンに変換します。
コマンド形式	`getregion('ip')` を入力します。
パラメーターの説明	IP アドレスです。

ツールバーで保存をクリックし、その後公開をクリックします。指示に従って、関数を開発環境および本番環境の MaxCompute プロジェクトに公開します。

データ処理ノードの構成

本チュートリアルでは、各変換レイヤーごとに MaxCompute SQL をスケジュールしてデータを処理します。dwd_log_info_di、dws_user_info_all_di、および ads_user_info_1d の各ノードに対して、提供されたサンプルコードを構成します。

dwd_log_info_di ノードの構成

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

DataStudio の左側ナビゲーションウィンドウでをクリックします。プロジェクトディレクトリ から作成済みのワークフローを見つけ、クリックしてワークフローキャンバスを開きます。
ワークフローキャンバス上で dwd_log_info_di ノードにカーソルを合わせ、ノードを開く をクリックします。

以下のコードをノードエディターに貼り付けます。

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 'リファラー URL',
 device STRING COMMENT 'デバイスの種別',
 identity STRING COMMENT 'アクセスの種別：クローラー、フィード、ユーザー、または不明'
)
PARTITIONED BY (
 dt STRING
)
LIFECYCLE 14;

-- データ処理
-- シナリオ：以下の SQL 文では、getregion 関数を使用して生のログデータから IP アドレスを解析します。
-- また、正規表現などの手法を用いて生データを分析可能なフィールドに分割し、dwd_log_info_di テーブルに書き込みます。
-- 注意：
--     1. DataWorks ノード内で関数を使用するには、まず必要なリソースをアップロードし、それらのリソースを用いて関数を登録する必要があります。
--        本チュートリアルでは、ip2region.jar リソースを用いて getregion 関数を登録します。
--     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 -- 正規表現を使用してリファラーをクリーニングし、より正確な URL を取得します。
  , CASE
    WHEN TOLOWER(agent) RLIKE 'android' THEN 'android' -- agent からデバイス情報およびアクセスの種別を取得します。
    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;

実行パラメーターを構成します。

MaxCompute SQL ノードエディターの右側ペインで、Run Configuration をクリックします。以下のパラメーターを構成します。これらのパラメーターは、「ステップ 4」でのテスト実行に使用されます。

パラメーター	説明
計算リソース	「環境の準備」の前段階でバインドした MaxCompute 計算リソースおよび対応するコンピューティングクォータを選択します。
リソースグループ	「環境の準備」の前段階で購入した Serverless リソースグループを選択します。
スクリプトパラメーター	構成は不要です。本チュートリアルのサンプルコードでは、ビジネス日付を表すために一律に `${bizdate}` を使用しています。「次のステップ」でワークフローを実行する際に、現在の値を `20250223` などの特定の定数に設定できます。タスクは実行時にこの定数で変数を置き換えます。

（任意）スケジューリング構成を構成します。
本チュートリアルでは、スケジューリングパラメーターのデフォルト値をそのまま使用できます。MaxCompute SQL ページの右側ペインで、スケジューリング をクリックします。スケジューリングパラメーターの詳細については、「ノードスケジューリング構成」をご参照ください。
- スケジューリングパラメーター：これはワークフロー全体でグローバルに構成されるため、個々のノードに対して構成する必要はありません。タスクやコード内で直接パラメーターを使用できます。
- スケジューリングポリシー：子ノードの実行をワークフロー開始後に遅延させるために、開始時刻 パラメーターを設定できます。本チュートリアルでは遅延を設定しません。
ツールバーで保存をクリックします。

dws_user_info_all_di ノードの構成

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

ワークフローキャンバス上で dws_user_info_all_di ノードにカーソルを合わせ、ノードを開く をクリックします。

以下のコードをノードエディターに貼り付けます。

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 'リファラー 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;

実行パラメーターを構成します。

パラメーター	説明
計算リソース	「環境の準備」の前段階でバインドした MaxCompute 計算リソースおよび対応するコンピューティングクォータを選択します。
リソースグループ	「環境の準備」の前段階で購入した Serverless リソースグループを選択します。
スクリプトパラメーター	構成は不要です。本チュートリアルのサンプルコードでは、ビジネス日付を表すために一律に `${bizdate}` を使用しています。「次のステップ」でワークフローを実行する際に、現在の値を `20250223` などの特定の定数に設定できます。タスクは実行時にこの定数で変数を置き換えます。

（任意）スケジューリング構成を構成します。
本チュートリアルでは、スケジューリングパラメーターのデフォルト値をそのまま使用できます。MaxCompute SQL ページの右側ペインで、スケジューリング をクリックします。スケジューリングパラメーターの詳細については、「ノードスケジューリング構成」をご参照ください。
- スケジューリングパラメーター：これはワークフロー全体でグローバルに構成されるため、個々のノードに対して構成する必要はありません。タスクやコード内で直接パラメーターを使用できます。
- スケジューリングポリシー：子ノードの実行をワークフロー開始後に遅延させるために、開始時刻 パラメーターを設定できます。本チュートリアルでは遅延を設定しません。
ツールバーで保存をクリックします。

ads_user_info_1d ノードの構成

このノードでは、dws_user_info_all_di テーブル内のデータを処理し、その結果を ads_user_info_1d テーブルに書き込んで基本的なユーザープロファイルを生成します。

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

以下のコードをノードエディターに貼り付けます。

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 'ページビュー (PV)',
 gender STRING COMMENT '性別',
 age_range STRING COMMENT '年齢層',
 zodiac STRING COMMENT '星座'
)
PARTITIONED BY (
 dt STRING
)
LIFECYCLE 14;    

-- データ処理
-- シナリオ：以下の SQL 文では、ワイドテーブル 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;

実行パラメーターを構成します。

パラメーター	説明
計算リソース	「環境の準備」の前段階でバインドした MaxCompute 計算リソースおよび対応するコンピューティングクォータを選択します。
リソースグループ	「環境の準備」の前段階で購入した Serverless リソースグループを選択します。
スクリプトパラメーター	構成は不要です。本チュートリアルのサンプルコードでは、ビジネス日付を表すために一律に `${bizdate}` を使用しています。「次のステップ」でワークフローを実行する際に、現在の値を `20250223` などの特定の定数に設定できます。タスクは実行時にこの定数で変数を置き換えます。

（任意）スケジューリング構成を構成します。
本チュートリアルでは、スケジューリングパラメーターのデフォルト値をそのまま使用できます。MaxCompute SQL ページの右側ペインで、スケジューリング をクリックします。スケジューリングパラメーターの詳細については、「ノードスケジューリング構成」をご参照ください。
- スケジューリングパラメーター：これはワークフロー全体でグローバルに構成されるため、個々のノードに対して構成する必要はありません。タスクやコード内で直接パラメーターを使用できます。
- スケジューリングポリシー：子ノードの実行をワークフロー開始後に遅延させるために、開始時刻 パラメーターを設定できます。本チュートリアルでは遅延を設定しません。
ツールバーで保存をクリックします。

データの処理

データを処理します。
ワークフローのツールバーで実行をクリックします。今回の実行で各ノードで定義されたパラメーター変数の値を設定します。本チュートリアルでは 20250223 を使用しますが、必要に応じて値を変更できます。その後、OK をクリックし、実行が完了するまで待ちます。
結果を確認します。
1. DataStudio の左側ナビゲーションウィンドウでをクリックして「データ開発」ページに移動します。その後、個人フォルダー領域でをクリックして、拡張子が .sql のファイルを作成します。ファイル名は任意に指定できます。
2. ページ下部で、言語モードが MaxCompute SQL であることを確認します。
3. SQL エディターウィンドウで、以下の SQL 文を実行して、最終テーブル ads_user_info_1d のレコード数を確認し、出力を検証します。
```
-- パーティション フィルターは、実際の操作のビジネス日付に変更する必要があります。
-- 本チュートリアルでは、'bizdate' パラメーターを 20250223 に設定しました。
SELECT count(*) FROM ads_user_info_1d WHERE dt='20250223';
```
  - コマンドが 0 より大きいカウントを返した場合、データ処理は完了しています。
  - データが返されない場合は、ワークフロー実行で構成された 今回の実行で使用する値 が、クエリ内の dt で指定されたデータタイムスタンプと一致しているか確認してください。ワークフローをクリックし、右側ペインで 実行履歴 をクリックした後、実行記録の アクション 列にある表示をクリックすると、ワークフロー実行ログ内のデータタイムスタンプ値 (partition=[pt=xxx]) を確認できます。

ワークフローの公開

タスクを本番環境に公開して、スケジュールに従って自動的に実行できるようにします。ワークフローの公開手順は以下のとおりです。

説明

本チュートリアルでは、「ワークフローのスケジューリング構成」で既にスケジューリングパラメーターを構成済みのため、公開前に各ノードに対して再度構成する必要はありません。

DataStudio の左側ナビゲーションウィンドウでをクリックします。プロジェクトディレクトリ から作成済みのワークフローを見つけ、クリックしてワークフローキャンバスを開きます。
ツールバーで公開をクリックして公開ダイアログボックスを開きます。
本番環境への公開を開始 をクリックします。表示される確認ダイアログボックスで、公開方法を選択します：
- フルリリース：現在のワークフローおよびその内部のすべてのタスクを公開します。
- インクリメンタルリリース：現在のワークフローおよび前回のリリース以降に変更されたノードのみを公開します。これは反復的な最適化や小規模な更新に適しています。
公開方法を確認後、システムが自動的にワークフローおよび選択されたノードを開発環境および本番環境にリリースします。処理を完了するには、プロンプトが表示されたら リリースの確認 をクリックします。

本番環境でのタスク実行

タスクを公開すると、翌日にインスタンスが生成されて実行されます。公開済みのワークフローを即座に実行して、本番環境で正常に動作することを検証するには、バックフィルデータ 機能を使用できます。詳細については、「データバックフィルインスタンス O&M」をご参照ください。

タスクが正常に公開された後、右上隅の オペレーションセンター をクリックします。
または、左上隅のアイコンをクリックし、すべてのプロダクト > データ開発および O&M > O&M (ワークフロー) を選択します。
左側ナビゲーションウィンドウで、周期タスク O&M > 定期タスク をクリックします。定期タスク ページで、workshop_start 仮想ノードをクリックします。
右側の DAG で、workshop_start ノードを右クリックし、バックフィルデータ > 現在および下流ノード を選択します。
バックフィルするタスクを選択し、ビジネス日付を設定して、送信してページへ移動 をクリックします。
バックフィルデータページで、更新をクリックして、すべての SQL タスクが成功するまで待ちます。

説明

チュートリアル完了後、課金されないように、ノードのスケジューリング有効期間を設定するか、ワークフローのルートノード (仮想ノード workshop_start) を[フリーズ]することができます。

次のステップ

データの可視化：ユーザープロファイル分析を完了した後、データ分析モジュールを使用して処理済みデータをチャートで可視化します。これにより、重要な情報を迅速に抽出し、ビジネスのトレンドに関する洞察を得ることができます。
データ品質のモニタリング：生成されたテーブルに対してデータ品質モニタリングルールを構成して、ダーティデータを早期に検出しブロックすることで、その影響が広がるのを防ぎます。
データの管理：ユーザープロファイル分析ワークフローが完了すると、対応するデータテーブルが MaxCompute に作成されます。これらのテーブルは Data Map モジュールで表示でき、データ系統を用いてテーブル間の関係を探索できます。
API を使用したデータサービスの提供：最終的な処理済みデータを取得した後、Data Service モジュールを使用して標準化された API を作成し、データを共有・活用することで、他のビジネスモジュールからの利用を可能にします。