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); -- ds でパーティション分割し、日次集計結果を保存

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

1. DataWorks コンソールの [ワークスペース] ページに移動します。上部のナビゲーションバーで、目的のリージョンを選択します。対象のワークスペースを見つけ、[操作] 列の [ショートカット] > [Data Studio] を選択します。

   ボタンのラベルが [データ開発] の場合、レガシー版の Data Studio が開きます。クリックしないでください。

2. 左側のナビゲーションバーで アイコンをクリックし、プロジェクトディレクトリ の右側にある > ワークフローの作成 をクリックして、ワークフローの作成 ページを開きます。

3. ワークフローの作成 ダイアログボックスで、スケジューリングタイプ を 定期スケジュール に設定し、必要な情報 (例：[名前] を minimal_daily_demo に設定) を入力して作成を完了します。

ステップ 3：ワークフローのオーケストレーション：ノードのドラッグと依存関係の接続

1. ワークフローキャンバスで、左側のコンポーネントパネルから仮想ノードをドラッグし、start_node と名付けます。

   ゼロ負荷ノードは、ビジネスプロセスの開始点を定義するためにのみ使用され、実際には実行されません。

2. MaxCompute SQL ノードをドラッグし、count_orders という名前を付けます。

3. start_node の下部にある円をクリックし、count_orders の上部まで線をドラッグして、単純な処理パイプラインを構築します。

ステップ 4：ノードコードの開発

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

2. ノードのビジネスロジックコードを記述します (ここではシミュレートされた統計データを使用します)。

   -- bizdate は定義済みのスケジューリング変数であり、その意味はスケジューリング構成で指定する必要があります
   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：スケジュールとパラメーターの構成

1. ワークフローに戻ります。ワークフローキャンバスの右側で、スケジューリング設定 > スケジュール時刻 タブをクリックします。

   1. スケジューリング周期 を 日 に設定します。

   2. スケジュール時刻 を 00:05 (つまり、毎日 00:05) に設定します。

2. count_orders ノードエディタの右側で、スケジューリング設定 > スケジューリングパラメーター を構成します。パラメーター名 を bizdate に、パラメーター値 を $[yyyymmdd-1] (現在の日付から1日を引いた日付、つまり前日を表します) に設定してパラメーターを追加します。

ステップ 6：単一ノードとワークフロー全体のデバッグ

1. count_orders ノードのデバッグ：

   1. デバッグパラメーターの構成：ノード編集ページの右側にある デバッグ設定 をクリックします。

      1. 計算リソース で、ステップ 1 で準備した MaxCompute 計算リソースを選択します。

      2. スクリプトパラメーター で、今回の実行値 を入力します。デフォルト値は現在の日付の前日です。

   2. デバッグタスクの実行：ツールバーの 実行 ボタンをクリックします。ノードは デバッグ設定 で構成したデバッグパラメーターで実行されます。

   3. 実行結果が期待通りになったら、右上隅の [スケジューリングに同期] をクリックして、実行構成をスケジュール設定に同期します。

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

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

   2. 表示されるダイアログボックスで、ワークフローの 今回の実行値 を入力します (例：今日が 20260120 の場合、bizdate は 20260119 に置き換える必要があります)。

ステップ 7：本番環境へのデプロイ

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

2. デプロイパネルで、システムが依存関係と構成のチェックを実行します。すべてが正しいことを確認した後、本番リリースの開始 をクリックし、デプロイ方法を フルリリース に設定します。

3. デプロイが成功したら、オペレーションセンター に移動し、ワークフローが定期タスクリストに表示されているか確認します。

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

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

スケジューリング依存関係は、スケジュール時刻の達成と上流の成功という2つの条件を通じて、独立したタスクを秩序あるデータ生成パイプラインに接続します。ワークフロー内でノードをオーケストレーションすると、スケジューリング依存関係が自動的に確立されます。また、スケジューリング依存関係の設定を通じて、より複雑な依存関係を構成することもできます。

詳細については、「スケジューリング依存関係の構成」をご参照ください。

パラメーターの設計とフロー

DataWorks Data Studio は4つのレベルのパラメーター渡しをサポートし、ノードコンテキストパラメーターを通じて柔軟なクロスノードの動的データ転送を可能にします。スコープが低い順にリストされています。

スケジュール設定

スケジュールはワークフローレベルで設定されます。内部ノードは遅延実行時間のみを設定でき、これはワークフローで定義されたスケジュール時刻に遅延時間を加算して計算されます。

単一ノードのデバッグ

単一ノードのデバッグは、SQL ステートメント、Python スクリプト、またはデータ統合同期タスクの検証など、1つのノード内のコードロジックを検証します。このモードは現在のノードのみを実行し、上流または下流の依存関係をトリガーしません。

1. ノードの右側にある デバッグ設定 パネルで、以下のパラメーターを構成します。

   パラメーター名	説明
   計算リソース	関連付けられた計算リソースを選択します。利用可能な計算リソースがない場合は、ドロップダウンリストから [計算リソースの作成] を選択します。 重要 計算リソースとリソースグループが接続されていることを確認してください。詳細については、「ネットワーク接続性」をご参照ください。
   リソースグループ	計算リソースが関連付けられた際に接続性テストに合格したリソースグループを選択します。一部のノードは、リソースグループに 依存パッケージを構成して実行環境を拡張することをサポートしています。
   (オプション) データセット	一部のノード (Shell や Python など) は、OSS や NAS に保存されている非構造化データにアクセスするためにデータセットをマウントすることをサポートしています。
   (オプション) スクリプトパラメーター	${parameter_name} 形式を使用してノードコンテンツを構成し、変数を定義する場合、スクリプトパラメーター で パラメーター名 と パラメーター値 を構成する必要があります。実行時に、変数は実際の値に動的に置き換えられます。詳細については、「スケジューリングパラメーターの構成」をご参照ください。
   (オプション) 関連付けられたロール	一部のノード (Shell や Python など) は、他の Alibaba Cloud サービスのリソースにアクセスするために関連付けられたロールを構成することをサポートしています。

2. ノードの上にあるツールバーで、保存 をクリックし、次に 実行 をクリックしてノードタスクを実行します。

3. ノードの下部でランタイムログと実行結果を表示します。

ワークフローのデバッグ

ワークフローのデバッグは、パイプライン内の複数のノードにまたがるデータ依存関係、パラメーターの受け渡し、実行順序が正しいかどうかを検証します。単一ノードのデバッグ後、部分的なパイプラインまたはワークフロー全体をデバッグできます。

1. ワークフローの DAG キャンバスページで、上部ツールバーの 実行 ボタンをクリックします。また、ノードを選択して右クリックし、このノードまで実行 または このノードから実行 を選択して部分的な検証を行うこともできます。

2. 表示されるダイアログボックスで、このデバッグ実行のためにワークフロー内のすべてのノード変数 (例：${bizdate}) に一時的な値を割り当てます。

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

4. キャンバスの左側にある 戻る ボタンをクリックして、ワークフローの開発状態に戻ることができます。

   右側の実行履歴には、ワークフローのすべてのデバッグ実行記録が表示されます。

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

DataWorks では、既存のスタンドアロンノードをワークフローにインポートしたり、ワークフローからノードを削除したり、ワークフロー間でノードを移動したりして、効率的な再利用とモジュラー管理が可能です。

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

どのワークフローにも属していない既存のスタンドアロンノードを、[ノードのインポート] を通じて現在のワークフローキャンバスに追加して再利用できます。

1. 対象のワークフローをダブルクリックして、キャンバス編集ページを開きます。

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

3. パネルには追加可能なすべてのスタンドアロンノードがリストされます。ノードタイプ、パス、または ノード名 でフィルタリングおよび検索して、対象のノードをすばやく見つけることができます。

4. ノードを見つけたら、キャンバスにドラッグしてインポートを完了します。

ワークフローからノードを削除する

ワークフローからノードを削除してスタンドアロンノードにしたり、直接別のワークフローに移動したりできます。

• スタンドアロンノードとしてワークフローから削除する

  ノードをワークフローから切り離し、ワークフローの一部でなくす必要がある場合に使用します。

  1. プロジェクトディレクトリツリーまたはワークフローキャンバスで、対象のノードを右クリックします。

  2. コンテキストメニューで、[ワークフローから削除] を選択します。

  3. 確認ダイアログで、削除後にノードが保存される対象パスを選択し、確認します。

• 別のワークフローに移動する

  ビジネスプロセスを再構築し、ノードを現在のワークフローから別のワークフローに移行する必要がある場合に使用します。

  1. プロジェクトディレクトリツリーまたはワークフローキャンバスで、対象のノードを右クリックします。

  2. コンテキストメニューで、[別のワークフローに移動] を選択します。

  3. 表示されるリストで、対象のワークフローを選択し、確認します。

     ノードを選択して対象のワークフローに直接ドラッグして移動することもできます。

ワークフローのクローン

クローンは、既存のワークフロー (すべての内部ノード、コード、依存関係を含む) をコピーして、独立したワークフローを生成します。クローンするには、プロジェクトディレクトリ に移動し、対象のワークフローを右クリックして クローン を選択します。

クローンはほぼすべての構成をコピーするため、リスクが生じる可能性があります。クローンしたワークフローをデプロイする前に、以下の点 (「環境分離」チェック) を確認してください。

1. 出力テーブルとターゲットデータソース：クローンされたワークフローのコードは、元のワークフローと同じターゲットテーブルに書き込みます。複数のタスクが同じテーブルを操作してデータ競合を引き起こすのを防ぐため、コード内のターゲットテーブル名を変更する (例：ods_user_table を dev_ods_user_table に変更) か、ノードの出力構成を変更する必要があります。

2. 上流依存関係：ワークフローの上流依存関係の構成を確認します。クローンされたワークフローは、デフォルトで元のの上流タスクに依存したままです。これが期待通りであるか確認してください。そうでない場合、データパイプラインの混乱やタスクのドライランが発生する可能性があります。

3. パラメーター構成：ワークフローとその内部ノードのカスタムパラメーター、特に日付パーティションに関連するパラメーター (例：${bizdate})、入出力パス、その他のパラメーターを確認します。新しい環境でこれらが正しいことを確認してください。

ワークフローのクローンに加えて、ワークフロー内のノードをクローンすることもできますが、ワークフローをまたいだノードのコピーはサポートされていません。

バージョン管理

バージョン管理は、すべてのワークフローの変更を自動的に記録し、履歴バージョンの表示と比較をサポートし、必要な場合には任意の履歴状態へのロールバックを可能にします。

ワークフローキャンバスの右側にある バージョン パネルには、2種類のレコードが表示されます。

• 開発記録：キャンバスで [保存] をクリックするたびに生成されます。このスナップショットは、開発中の偶発的なコード損失を防ぎ、本番環境で実行中のタスクには影響しません。

• デプロイ記録：ワークフローが本番環境にデプロイされた後に生成されます。これは、本番スケジューラが実行するバージョンです。本番タスクで問題が発生した場合は、ロールバックのためにデプロイ記録に注目してください。

典型的なユースケース

• 障害時の迅速なロールバック：コード変更により本番タスクが失敗した場合、[復元] を使用してワンクリックでワークフローを最後の安定したデプロイバージョンに戻し、ダウンタイムを最小限に抑えます。

• 変更の監査と追跡：いつ、誰によってロジックが変更されたかを調査するには、バージョンリストと [差分] 機能を使用して各コード変更を追跡します。

• コードの回復：保存せずに誤ってコードロジックを削除してしまった場合、最新の開発記録から復元します。

重要な考慮事項

• 復元は現在の開発エリアを上書きします：[復元] は、選択した履歴バージョンを使用して、現在のキャンバス上のすべてのコードと構成を上書きします。復元する前に、コミットされていない変更をローカルのテキストエディタにコピーしてバックアップしてください。

• 復元後には再デプロイが必要です：[復元] はコードを開発状態に戻すだけです。ロールバックを本番環境で有効にするには、手動でデプロイする必要があります。

デプロイと運用

• ノード/ワークフローのデプロイ：ワークフロー開発を完了した後、ノード/ワークフローを本番環境にデプロイします。この時点で、開発環境のノードは、本番環境に対応する定期タスクを生成します。詳細については、「ノードとワークフローのデプロイ」をご参照ください。

• ノード/ワークフローの運用：本番環境のワークフローは、スケジュール設定に基づいて定期的にスケジューリングされます。オペレーションセンター に移動して、定期実行ワークフローのスケジューリングステータスを表示し、関連する操作を実行します。詳細については、「定期タスク」および「定期実行インスタンス」をご参照ください。

タスク実行メカニズム

スケジューリングシナリオでは、ワークフロー全体の成功は、その内部タスクの実行ステータスに依存します。

ワークフロー全体の成功は、その内部ノードの最終ステータスによって決まります。

• 失敗：いずれかの重要なノードが失敗した場合、通常、ワークフローインスタンス全体が失敗とマークされます。

• 凍結/一時停止：ワークフロー内のノードが手動で凍結または一時停止され、そのノードが後続ノードの上流ノードである場合、パイプラインは中断され、ワークフロー全体も失敗とマークされます。

• 結合ノード：ワークフローが結合ノードを使用している場合、上流の分岐のノードが失敗しても、結合ノード自体のロジックが成功と判断すれば、ワークフロー全体が「成功」を返すことがあります。したがって、タスクがエラーを検出されずに実行されるのを防ぐために、結合ノードの上流チェックを慎重に設計してください。

特殊なシナリオ：

• ワークフロータスクのバックフィルデータインスタンスを凍結すると、ワークフローインスタンスは成功ステータスに設定されます。

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