DataWorks:代入ノード

例 1：MaxCompute SQL クエリ結果の受け渡し

例 2：Python 2 の出力結果の受け渡し

Python 2 の print ステートメントの出力は、カンマ (,) で分割され、1次元配列として下流ノードに渡されます。

• 上流ノード (代入ノード - Python 2) の設定

  Python 2 のコードは次のとおりです：

  print 'Electronics, Clothing, Books';

• 下流ノード (シェルノード) の設定と出力

  シェルノードで、上流の代入ノードの outputs を参照する types という名前の入力パラメーターを追加します。

  データを読み込むために次のコードを記述します： # 1次元配列全体を出力 echo "Full result set: ${types}" # インデックスで配列から要素を出力 echo "Second element: ${types[1]}"

  DataWorks はパラメーターを直接解析し、静的な置換を実行します。実行時の出力は次のようになります： Full result set: Electronics, Clothing, Books Second element: Clothing

背景情報

ある大手インターネット企業でデータエンジニアとして、E コマース (ecom)、金融 (finance)、物流 (logistics) という 3 つの中核ビジネスラインのデータ処理を担当しているとします。将来的には、さらに多くのビジネスラインが追加される可能性があります。これらのビジネスラインのユーザー行動ログに対して、毎日同じ集約ロジックを実行し、各ユーザーの PV (ページビュー) を計算し、その結果を共通の集計テーブルに保存する必要があります。

• 上流のソーステーブル (DWD レイヤー)：

  • dwd_user_behavior_ecom_d：E コマースユーザー行動テーブル。

  • dwd_user_behavior_finance_d：金融ユーザー行動テーブル。

  • dwd_user_behavior_logistics_d：物流ユーザー行動テーブル。

  • dwd_user_behavior_${biz_line}_d：他の潜在的なビジネスラインのユーザー行動テーブル。

  • これらのテーブルは同じ構造を持ち、日 (dt) でパーティション分割されています。

• 下流のターゲットテーブル (DWS レイヤー)：

  • dws_user_summary_d：ユーザーサマリーテーブル。

  • このテーブルは、すべてのビジネスラインからの集約結果を保存するために、ビジネスライン (biz_line) と日 (dt) でパーティション分割されています。

ビジネスラインごとに個別のタスクを作成すると、メンテナンスが煩雑になり、エラーが発生しやすくなります。For-Each ノードを使用すると、1つの処理ロジックを維持するだけで、システムがすべてのビジネスラインを自動的に反復処理して計算を完了するため、メンテナンスが簡素化されます。

データ準備

まず、サンプルテーブルを作成し、テストデータを挿入します。この例では、業務日として 20251010 を使用します。

1. ワークスペースに MaxCompute コンピュートエンジンをバインドします。

2. Data Studio に移動し、MaxCompute SQL ノードを作成します。

3. ソーステーブル (DWD レイヤー) を作成します：以下のコードを MaxCompute SQL ノードに追加し、選択して実行します。

   -- E コマースユーザー行動テーブル
   CREATE TABLE IF NOT EXISTS dwd_user_behavior_ecom_d (
       user_id     STRING COMMENT 'ユーザー ID',
       action_type STRING COMMENT 'アクションタイプ',
       event_time  BIGINT COMMENT 'イベントの UNIX タイムスタンプ (ミリ秒)'
   ) 
   COMMENT 'E コマースユーザー行動詳細ログテーブル'
   PARTITIONED BY (dt STRING COMMENT '日付パーティション、形式 yyyymmdd');
   INSERT OVERWRITE TABLE dwd_user_behavior_ecom_d PARTITION (dt='20251010') VALUES
   ('user001', 'click',        1760004060000), -- 2025-10-10 10:01:00.000
   ('user002', 'browse',       1760004150000), -- 2025-10-10 10:02:30.000
   ('user001', 'add_to_cart',  1760004300000); -- 2025-10-10 10:05:00.000
   -- E コマースユーザー行動テーブルが作成されたことを確認します。
   SELECT * FROM dwd_user_behavior_ecom_d where dt='20251010';
   -- 金融ユーザー行動テーブル
   CREATE TABLE IF NOT EXISTS dwd_user_behavior_finance_d (
       user_id     STRING COMMENT 'ユーザー ID',
       action_type STRING COMMENT 'アクションタイプ',
       event_time  BIGINT COMMENT 'イベントの UNIX タイムスタンプ (ミリ秒)'
   ) 
   COMMENT '金融ユーザー行動詳細ログテーブル'
   PARTITIONED BY (dt STRING COMMENT '日付パーティション、形式 yyyymmdd');
   INSERT OVERWRITE TABLE dwd_user_behavior_finance_d PARTITION (dt='20251010') VALUES
   ('user003', 'open_app',      1760020200000), -- 2025-10-10 14:30:00.000
   ('user003', 'transfer',      1760020215000), -- 2025-10-10 14:30:15.000
   ('user003', 'check_balance', 1760020245000), -- 2025-10-10 14:30:45.000
   ('user004', 'open_app',      1760020300000); -- 2025-10-10 14:31:40.000
   -- 金融ユーザー行動テーブルが作成されたことを確認します。
   SELECT * FROM dwd_user_behavior_finance_d where dt='20251010';
   -- 物流ユーザー行動テーブル
   CREATE TABLE IF NOT EXISTS dwd_user_behavior_logistics_d (
       user_id     STRING COMMENT 'ユーザー ID',
       action_type STRING COMMENT 'アクションタイプ',
       event_time  BIGINT COMMENT 'イベントの UNIX タイムスタンプ (ミリ秒)'
   ) 
   COMMENT '物流ユーザー行動詳細ログテーブル'
   PARTITIONED BY (dt STRING COMMENT '日付パーティション、形式 yyyymmdd');
   INSERT OVERWRITE TABLE dwd_user_behavior_logistics_d PARTITION (dt='20251010') VALUES
   ('user001', 'check_status',    1760032800000), -- 2025-10-10 18:00:00.000
   ('user005', 'schedule_pickup', 1760032920000); -- 2025-10-10 18:02:00.000
   -- 物流ユーザー行動テーブルが作成されたことを確認します。
   SELECT * FROM dwd_user_behavior_logistics_d where dt='20251010';

4. ターゲットテーブル (DWS レイヤー) を作成します：以下のコードを MaxCompute SQL ノードに追加し、選択して実行します。

   CREATE TABLE IF NOT EXISTS dws_user_summary_d (
       user_id     STRING COMMENT 'ユーザー ID',
       pv          BIGINT COMMENT 'PV'
   ) 
   COMMENT 'ユーザー日次 PV サマリーテーブル'
   PARTITIONED BY (
       dt           STRING COMMENT '日付パーティション、形式 yyyymmdd',
       biz_line     STRING COMMENT 'ビジネスラインパーティション (ecom、finance、logistics など)'
   );

   重要

   ワークスペースが Standard Edition の場合、このノードを本番環境にデプロイし、データバックフィルを実行する必要があります。

ワークフローの実装

1. ワークフローを作成します。右側の [スケジューリングパラメーター] パネルで、スケジューリングパラメーター bizdate を前日に設定します：$[yyyymmdd-1]。

2. ワークフローで、get_biz_list という名前の代入ノードを作成し、MaxCompute SQL 言語を使用して以下のコードを記述します。このノードは、処理対象のビジネスラインのリストを出力します：

   -- 処理対象のすべてのビジネスラインを出力
   SELECT 'ecom' AS biz_line
   UNION ALL
   SELECT 'finance' AS biz_line
   UNION ALL
   SELECT 'logistics' AS biz_line;

3. For-Each ノードの設定

   • ワークフローキャンバスに戻り、get_biz_list 代入ノードの下流に For-Each ノードを作成します。

   • For-Each ノードの設定ページに移動します。右側のスケジューリング設定パネルの Scheduling Parameters > Script Parameters で、loopDataArray パラメーターを get_biz_list ノードの outputs にバインドします。

   • For-Each ノードのループ本体で、[内部ノードの作成] をクリックして MaxCompute SQL ノードを作成し、ループ本体の処理ロジックを記述します。

     説明

     • このスクリプトは For-Each ノードによって駆動され、ビジネスラインごとに 1 回実行されます。

     • 実行時、組み込み変数 ${dag.foreach.current} は、'ecom'、'finance'、'logistics' といった現在のビジネスライン名に動的に置換されます。

     SET odps.sql.allow.dynamic.partition=true;
     INSERT OVERWRITE TABLE dws_user_summary_d PARTITION (dt='${bizdate}', biz_line)
     SELECT
         user_id,
         COUNT(*) AS pv,
         '${dag.foreach.current}' AS biz_line
     FROM
         dwd_user_behavior_${dag.foreach.current}_d
     WHERE
         dt = '${bizdate}'
     GROUP BY
         user_id;

4. 検証ノードの追加

   ワークフローに戻ります。For-Each ノードの [下流ノードの作成] をクリックして MaxCompute SQL ノードを作成し、以下のコードを追加します。

   SELECT * FROM dws_user_summary_d WHERE dt='20251010' ORDER BY biz_line, user_id;

デプロイと結果

ワークフローを本番環境にデプロイします。Auto Triggered Task O&M > Auto Triggered Task に移動し、ターゲットのワークフローを見つけて、業務日を '20251010' に設定してスモークテストを実行します。

実行が完了したら、テストインスタンスで実行ログを表示します。最終ノードは以下の出力を生成します：

メリット

• 高いスケーラビリティ：新しいビジネスラインを追加するには、代入ノードに SQL を 1 行追加するだけで済みます。処理ロジックの変更は不要です。

• 容易なメンテナンス：すべてのビジネスラインが同じ処理ロジックを共有します。1 か所の変更がすべてに反映されます。