DataWorks:ワークフロー

ステップ 1：コンピュートエンジンと結果テーブルの準備

このステップでは、ワークフローが利用する基盤インフラストラクチャを設定します。

1. 対象のワークスペースで、MaxCompute をコンピュートエンジンとしてバインドします。

2. MaxCompute で結果テーブルを作成します。

   -- 日次注文数を格納するための結果テーブルを作成
   CREATE TABLE IF NOT EXISTS dw_order_count_test (
       order_date STRING,
       total_count BIGINT
   )
   PARTITIONED BY (ds STRING); -- 日付によるパーティション

ステップ 2：定期実行ワークフローの作成

このステップでは、すべてのノードとスケジューリング構成を保持するワークフローのコンテナを作成します。

1. DataWorks コンソールの「ワークスペース」ページに移動します。上部のナビゲーションバーで、目的のリージョンを選択します。対象のワークスペースを見つけ、「ショートカット」＞「Data Studio」を「操作」列から選択します。＞ ボタンに「Data Development」と表示されている場合は、旧バージョンが開きます。クリックしないでください。

2. 左側のナビゲーションウィンドウにある アイコンをクリックします。「ワークスペースディレクトリ」の右側にある  をクリックし、「ワークフローの作成」を選択します。

3. 「新規ワークフロー」ダイアログボックスで、「スケジューリングタイプ」を「定期実行スケジューリング」に設定し、名前（例： minimal_daily_demo）を入力して作成を完了します。

ステップ 3：ワークフローのオーケストレーション

このステップでは、DAG キャンバスを使用してパイプラインの構造を定義します。

1. ワークフローキャンバス上で、左側のコンポーネントパネルから 仮想ノード をドラッグし、名前を start_node に設定します。＞ 仮想ノード はワークフローの開始ポイントを定義するプレースホルダーであり、実際の計算処理は行いません。

2. 「MaxCompute SQL」ノードをキャンバス上にドラッグし、名前を count_orders に設定します。

3. start_node の下部にあるドットをクリックし、count_orders の上部まで線をドラッグして依存関係を作成します。

ステップ 4：ノードコードの記述

このステップでは、毎日実行されるビジネスロジックを追加します。

1. count_orders をダブルクリックして、ノードのコードエディタを開きます。

2. SQL ロジックを記述します。この例では、シミュレートされたデータを使用します。

   -- bizdate は、ランタイム時に注入されるスケジューリングパラメーターです。
   -- $[yyyymmdd-1] は前日の日付（例：本日が 20260120 の場合、20260119）に解決されます。
   INSERT OVERWRITE TABLE dw_order_count_test PARTITION (ds='${bizdate}')
   SELECT
       '${bizdate}' AS order_date,
       COUNT(*) AS total_count
   FROM (SELECT 1 AS id UNION ALL SELECT 2 AS id) t; -- シミュレートされたデータ

   ＞ ノード開発の詳細については、「MaxCompute SQL ノード」をご参照ください。

3. 保存 をクリックします。

ステップ 5：スケジューリングサイクルとパラメーターの構成

このステップでは、ワークフローの実行タイミングと、SQL に渡す日付値を指定します。

1. ワークフローキャンバスに戻ります。右側のパネルで、「スケジューリング」＞「スケジュール時刻」に移動します。

   • 「スケジューリングサイクル」を「毎日」に設定します。

   • 「スケジュール時刻」を 00:05 に設定します。

2. count_orders ノードエディタで、「スケジューリング構成」＞「スケジューリングパラメーター」に移動し、以下を追加します。

   • 「パラメーター名」：bizdate

   • 「パラメーター値」：$[yyyymmdd-1]（前日の日付）

ステップ 6：ノードおよびワークフローのデバッグ

デバッグは、ワークフローを本番環境に投入する前に、コードロジックを検証するためのものです。

1. ノードのデバッグ

   1. ノードエディタの右側で、「デバッグ構成」をクリックします。

      • 「コンピュートエンジン」で、ステップ 1 で設定した MaxCompute エンジンを選択します。

      • 「スクリプトパラメーター」で、「今回の実行で使用する値」（デフォルトは前日の日付）を設定します。

   2. ツールバーの「実行」をクリックします。ノードは、指定したパラメーターで実行されます。

   3. 結果が正しく表示されたら、「スケジューリング構成への同期」をクリックして、デバッグ構成をスケジューリング構成に適用します。

2. ワークフロー全体のデバッグ

   1. ワークフローキャンバスの上部ツールバーにある アイコンをクリックします。

   2. ダイアログボックスで、ワークフロー変数の「現在の実行値」を入力します。たとえば、本日が 20260120 の場合、bizdate を 20260119 に設定します。

ステップ 7：本番環境への公開

公開を行うと、定義されたスケジュールで実行される定期実行タスクが本番環境に作成されます。

1. ワークフローに戻り、上部ツールバーの ボタンをクリックします。

2. 公開パネルで、システムが依存関係および構成のチェックを実行します。チェックが完了したら、「本番環境へのデプロイメントを開始」をクリックし、「完全デプロイメント」を選択します。

3. 「オペレーションセンター」に移動し、ワークフローが定期実行タスクリストに表示されることを確認します。

これで、ワークフローは毎日 00:05 に実行されるようになります。

ワークフローのオーケストレーション

ロジック再利用のためのサブワークフロー

データクレンジング、要約統計、品質チェックなど、複数のワークフローが共通の処理ステップを共有する場合、それらを サブワークフロー として再利用可能にカプセル化し、SUB_PROCESS ノードを使用します。これにより、ワークフロー間でのロジック重複を回避できます。共有ステップが変更された場合、サブワークフローのみを更新すれば済みます。

サブワークフローの作成および使用方法：

1. 「child_workflow」ワークフローで、「プロパティ」を「参照可能」に設定します。ワークフローがサブワークフローになります。

2. メインワークフローで、SUB_PROCESS ノードをドラッグし、「child_workflow」を参照先ワークフローとして選択します。

サブワークフローには以下の制約があります。

• 内部専用の依存関係：サブワークフローおよびその内部ノードは、外部タスクに依存できません。

• 分離性：サブワークフローは、外部タスクの直接的な依存関係になることはできません。

• 受動的トリガー：公開後は、スケジュールインスタンスが自動生成されません。SUB_PROCESS ノードから呼び出された場合にのみ実行されます。＞ 詳細については、「SUB_PROCESS ノード」をご参照ください。

スケジューリング依存関係

スケジューリング依存関係は、孤立したノードを順序立てられたデータパイプラインに接続します。ノードは、スケジュールされた時刻が到来したうえで、すべての上流タスクが正常終了した場合にのみ実行されます。

完全なリファレンスについては、「スケジューリング依存関係」をご参照ください。

キャンバス上でノードを接続すると、DataWorks が自動的にそれらの間にスケジューリング依存関係を作成します。より複雑なクロスワークフローまたはクロスサイクルの依存関係については、以下の依存関係タイプを構成します。

パラメーター設計

ワークスペースパラメーターは、DataWorks Professional Edition 以降でのみ利用可能です。

DataWorks は、範囲が狭い順から広い順に並べた 4 段階のパラメーター体系をサポートしています。ノード間で構成を重複させないために、適切な段階を選択してください。

スケジューリングサイクル

スケジューリングサイクルはワークフローレベルで設定されます。ワークフロー内のノードは、独自の独立したサイクルを持つことはできず、ワークフローのサイクルを継承し、ワークフローのスケジュール時刻からのオフセットとして「遅延実行時刻」のみを指定できます。

単一ノードのデバッグ

単一ノードのデバッグは、上流／下流の依存関係を一切トリガーせずに、SQL ステートメント、Python スクリプト、データ統合タスクなどのノード内部ロジックを検証します。

1. ノードの右側パネルで、「デバッグ構成」を構成します。

   パラメーター	説明
   コンピュートエンジン	バインド済みのコンピュートエンジンを選択します。利用可能なエンジンがない場合は、ドロップダウンリストから「コンピュートエンジンの作成」を選択します。コンピュートエンジンとリソースグループが接続されていることを確認してください。「ネットワーク接続ソリューション」をご参照ください。
   リソースグループ	コンピュートエンジンのバインド時に接続性テストを通過したリソースグループを選択します。一部のノードでは、リソースグループにカスタムイメージを適用することで、実行環境を拡張できます。
   データセット（任意）	Shell および Python ノードは、データセットの使用をサポートしており、OSS 内の非構造化データにアクセスできます。
   スクリプトパラメーター（任意）	ノードコード内で ${parameter_name} を使用して変数を定義します。ここでは「パラメーター名」と「パラメーター値」を設定します。「スケジューリングパラメーターのソースと式」をご参照ください。
   関連付けロール（任意）	Shell および Python ノードは、関連付けロールの構成をサポートしており、他のクラウドプロダクトのリソースにアクセスできます。

2. 保存 をクリックし、続いて 実行 をクリックします。

3. ノードエディタの下部で実行ログおよび結果を確認します。

ワークフローのデバッグ

ワークフローのデバッグは、データ依存関係、パラメーターの受け渡し、およびパイプライン全体の実行順序を検証します。個々のノードのデバッグを終えた後、ワークフロー全体または部分的なパイプラインをデバッグします。

1. ワークフローの DAG キャンバスで、上部ツールバーの 実行 をクリックします。部分的なパイプラインを検証するには、ノードを右クリックし、「このノードまで実行」または「このノードから実行」を選択します。

2. ダイアログボックスで、このデバッグセッション用にすべてのノード変数（例：${bizdate}）に一時的な値を割り当てます。

3. システムは、DAG で定義された依存関係に従って、上から下へ順次ノードを実行します。キャンバス上でノードのステータスをリアルタイムで監視し、任意のノードをクリックして実行ログを表示できます。

4. キャンバス左側の 戻る をクリックして、開発状態に戻ります。右側の実行履歴には、すべてのデバッグ実行記録が表示されます。

ワークフローノードの管理

DataWorks では、スタンドアロンノードをワークフローにインポートしたり、ノードをワークフローから移動したり、ワークフロー間でノードを転送したりできます。これらは、ノードをゼロから再構築することなく実行可能です。

既存ノードをワークフローにインポート

1. 対象のワークフローをダブルクリックして、キャンバスエディタを開きます。

2. 左側のコンポーネントパネルで、「ノードのインポート」タブに切り替えます。

3. 「ノードタイプ」、「パス」、または「ノード名」でフィルターをかけて、対象ノードを探します。

4. ノードをキャンバス上にドラッグします。

ノードをワークフローから移動

• スタンドアロンノードへの変換：ディレクトリツリーまたはキャンバス上のノードを右クリックし、「ワークフローから移動」を選択し、対象パスを選んで確定します。

• 別のワークフローへの移動：ノードを右クリックし、「別のワークフローに移動」を選択し、対象ワークフローを選んで確定します。また、ノードを直接対象ワークフローにドラッグすることもできます。

ワークフローのクローン作成

ワークフローのクローン作成は、すべての内部ノード、コード、依存関係を含む完全かつ独立したコピーを作成します。「ワークスペースディレクトリ」で対象のワークフローを右クリックし、「クローン」を選択します。

クローン作成したワークフローをコミットおよび公開する前に、以下の点を確認して、データ競合およびパイプラインの問題を防止してください。

1. 出力テーブルおよび対象データソース：クローン作成したワークフローは、元のワークフローと同じ対象テーブルに書き込みます。複数のタスクが同一テーブルに書き込まないように、コード内のテーブル名を更新します（例： ods_user_table から dev_ods_user_table へ）。

2. 上流依存関係：クローン作成したワークフローは、デフォルトで元の上流タスクに依存したままです。これが意図した動作かどうかを確認してください。

3. パラメーター構成：日付パーティション（例：${bizdate}）や入出力パスを含むカスタムパラメーターを確認し、新しい環境で正しく設定されていることを確認します。

ワークフロー内の個々のノードをクローン作成することもできますが、異なるワークフロー間でノードをコピーすることはできません。

バージョン管理

バージョン管理は、ワークフローに対するすべての変更を自動的に記録します。「ワークフローキャンバス」右側の「バージョン」パネルから、任意の以前のバージョンを表示、比較、復元できます。

バージョンパネルには、以下の 2 種類の記録が表示されます。

• 開発記録：「保存」をクリックするたびに作成されます。これは進行中のスナップショットであり、本番環境には影響しません。

• 公開記録：ワークフローが本番環境に正常に公開されたときに作成されます。これは、本番スケジューリングシステムが実行するものです。

代表的なユースケース

復元前の注意事項

• 復元は現在の開発領域を上書きします：キャンバス上のすべてのコードおよび構成が、履歴バージョンで置き換えられます。未保存の変更がある場合は、復元前にローカルでバックアップしてください。

• 復元後に再公開する必要があります：復元は開発状態のみを復元します。ロールバックを本番環境で有効にするには、ワークフローを手動で公開してください。

公開および運用

• 公開：ワークフローの開発が完了したら、本番環境に公開します。開発環境のノードは、本番環境に対応する定期実行タスクを生成します。「ノードまたはワークフローの公開」をご参照ください。

• 運用：ワークフローは、スケジューリング構成に基づいて定期的に実行されます。「オペレーションセンター」に移動して、スケジューリングステータスを監視およびタスクを管理します。「定期実行タスクの基本的な運用タスク」および「Data Backfill インスタンスの運用タスク」をご参照ください。

ワークフローインスタンスのステータス

ワークフローの全体的なステータスは、その内部ノードの最終ステータスを反映します。システムがワークフローのステータスをどのように決定するかを理解することで、問題の診断を迅速化できます。

特殊なシナリオ：

• ワークフロータスクの Data Backfill インスタンスをフリーズすると、ワークフローインスタンスは 成功 と設定されます。

• Data Backfill シナリオにおいて、システムがタスクを実行できないと判断した場合、ワークフローは 失敗 と設定されます。