DataWorks:for-each ノード

組み込みパラメーター

for-each ループ本体内のノードは、以下の組み込み変数を使用してループのステータスとデータにアクセスできます：

for(int i=0;i<data.length;i++) {
   print(data[i]);
}

上流の出力が SQL クエリの結果のような2 次元配列である場合、次の構文を使用して特定の値にアクセスすることもできます：

シナリオ 1：1 次元配列の処理

• 代入ノードの出力：2025-11-01,2025-11-02,2025-11-03

• 反復回数：3

• 2 回目の反復処理中：

  • ${dag.foreach.current} の値は 2025-11-02 です。

  • ${dag.loopTimes} の値は 2 です。

シナリオ 2：2 次元配列の処理

• 代入ノード (MaxCompute SQL) の出力：

  +-----+----------+
  | id  | city     |
  +-----+----------+
  | 101 | beijing  |
  | 102 | shanghai |
  +-----+----------+

• 反復回数：2

• 2 回目の反復処理中：

  • ${dag.foreach.current} の値は 102,shanghai です。

  • ${dag.loopTimes} の値は 2 です。

  • ${dag.foreach.current[0]} の値は 102 です。

  • ${dag.foreach.current[1]} の値は shanghai です。

背景情報

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

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

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

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

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

  • dwd_user_behavior_${line_of_business}_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 '日次ページビュー'
   ) 
   COMMENT 'ユーザー日次ページビュー集計テーブル'
   PARTITIONED BY (
       dt           STRING COMMENT '日付パーティション (yyyymmdd 形式)',
       biz_line     STRING COMMENT '業務ラインパーティション (ecom, finance, logistics など)'
   );

   重要

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

ワークフローの実装

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 ノードによって駆動され、各業務ラインに対して一度実行されます。

     • ランタイム時に、システムは組み込み変数 ${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 O&M > Auto Triggered Task に移動し、ワークフローを見つけてスモークテストを実行します。ビジネス日付として '20251010' を選択します。

実行が完了すると、最終ノードは次の出力を生成します：

メリット

• 高いスケーラビリティ：新しい業務ラインを追加するには、処理ロジックを変更することなく、代入ノードに 1 行の SQL を追加するだけです。

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