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 コンソールの[ワークスペース] ページに移動します。 上部のナビゲーションバーで目的のリージョンを選択し、目的のワークスペースを見つけて、[アクション] 列で [ショートカット] > [DataStudio]を選択します。

   ボタンに「データ開発」と表示されている場合は、古いバージョンが開きます。クリックしないでください。

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. デプロイメントが成功したら、[オペレーションセンター] に移動して、ワークフローが定期タスクリストに表示されるかどうかを確認します。

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

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

スケジューリング依存関係は、データ処理の順序、一貫性、精度を保証するためのコアメカニズムです。スケジュールされた時刻に到達することと、上流タスクの成功によってトリガーされるというデュアルロジックを通じて、孤立したタスクを秩序あるデータ生産ラインに接続します。ワークフロー内でノードをオーケストレーションすると、それらの間にデフォルトでスケジューリング依存関係が確立されます。ワークフロー内のノードの上流および下流の依存関係に加えて、より複雑な依存関係を設定できます。

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

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

新しい Data Studio は、4 段階のパラメーター渡しメカニズムをサポートし、コンテキストパラメーターを通じてノード間で柔軟かつ動的なデータ渡しを可能にします。スコープが狭い順から広い順に、レベルは次のようになります：

スケジューリング周期

スケジューリング周期はワークフローレベルでのみ設定できます。ワークフロー内のノードは、ワークフローのスケジュール時刻からのオフセットである遅延実行時間のみを持つことができます。

単一ノードのデバッグ

単一ノードのデバッグは、SQL ステートメント、Python スクリプト、またはデータ統合タスクを迅速に検証するなど、単一ノードの内部コードロジックの検証に焦点を当てます。このモードでは現在のノードのみが実行され、上流または下流の依存関係はトリガーされません。

1. ノード右側の[実行設定] パネルで、以下のパラメーターを設定します。

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

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 種類のレコードが表示され、それらの違いを理解することが重要です。

• 開発レコード：キャンバスで [保存] をクリックするたびに生成されます。これらは単に開発プロセスのスナップショットであり、誤ってコードを失うのを防ぐためのものです。本番環境で現在実行中のタスクには影響しません。

• デプロイレコード：ワークフローが本番環境に正常に公開されたときに生成されます。これは、オンラインのスケジューリングシステムが実行するバージョンです。本番タスクで問題が発生した場合は、ロールバック時にデプロイレコードに注目する必要があります。

典型的なユースケース

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

• 変更の監査とトレーサビリティ：特定のロジックがいつ、誰によって変更されたかを調査する必要がある場合、[バージョンリスト] と [比較] 機能を使用して、すべてのコード変更を明確に追跡できます。

• コードの回復：誤って一部のコードロジックを削除し、保存していなかった場合、最新の開発レコードにロールバックすることで回復できます。

重要な注意点

• ロールバックは現在の開発ワークスペースを上書きします：ロールバック操作は、現在のキャンバス上のすべてのコードと設定を選択した履歴バージョンの内容で直接上書きします。現在のキャンバスにコミットされていない有効な変更がある場合は、ロールバックする前に、手動でローカルのテキストエディタにコピーしてバックアップすることを強く推奨します。

• ロールバック後に再公開する必要があります：ロールバック操作は、コードを開発状態に戻すだけです。この復元されたバージョンは、自動的に本番環境に同期されません。変更を本番環境で有効にするには、ワークフローを再公開する必要があります。

デプロイと運用

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

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

タスク実行メカニズム

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

ワークフロー全体の成功は、以下のルールに従って、その内部ノードの最終状態によって決定されます：

• 失敗：単一のクリティカルなノードが失敗した場合、通常、ワークフローインスタンス全体が失敗に設定されます。

• フリーズ/一時停止：ワークフロー内のノードが手動でフリーズまたは一時停止され、そのノードが後続ノードの上流依存関係である場合、パイプラインは中断され、ワークフロー全体も失敗に設定されます。

• 結合ノード：ワークフローが結合ノードを使用する場合、結合ノードのロジックが成功を許容する限り、一部の上流ブランチノードが失敗しても、ワークフロー全体は「成功」を返すことがあります。したがって、「エラーありで実行」を防ぐために、結合ノードの上流チェックを慎重に設計する必要があります。

特別なシナリオ：

• ワークフロータスクのデータ補完インスタンスをフリーズすると、ワークフローインスタンスは成功状態に設定されます。

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