DataWorks:トリガーワークフロー

ステップ 1：トリガーワークフローの作成

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

2. 左側ナビゲーションバーで をクリックし、プロジェクトフォルダー の右側にある > 新しいワークフロー をクリックして、新しいワークフロー ページに移動します。

3. 表示されたダイアログボックスで、ワークフローの作成 ページにて、スケジューリングタイプ を トリガー方式スケジューリング に設定します。名前 を入力し、確認 をクリックします。

ステップ 2：ノードのオーケストレーションと開発

1. ツールバーの + ノードを追加 をクリックしてノード一覧を開き、ノードタイプ一覧から Shell ノードをキャンバスにドラッグ＆ドロップして名前を入力します。

2. Shell ノードをダブルクリックしてコードエディタを開き、以下のコードを入力します：

   echo "Hello, Trigger Workflow! Current time is ${bizdate}"

3. ツールバーの 保存 をクリックします。

ステップ 3：デバッグおよび実行（開発環境）

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

2. 表示されたダイアログボックスで、ワークフローの 今回の実行値 を入力します（例：現在日付が 20260310 の場合、bizdate の値は 20260309 となります）。

3. すぐに下部の実行時ログに、ノードの実行状況および echo コマンドの出力が表示されます。

ステップ 4：公開および実行（本番環境）

1. ワークフローキャンバスで 公開 ボタンをクリックし、プロンプトに従って公開します。

2. ワークフローが公開された後、オペレーションセンター > ワンタイムタスク O&M > ワンタイムタスク > トリガーワークフロー に移動します。

3. 公開済みのワークフローを見つけ、実行 を 操作 列からクリックします。

4. 表示されたウィンドウで、再度 実行 をクリックして、本番環境でワークフローのインスタンスを起動します。この実行の詳細は ワンタイムインスタンス ページで確認できます。

これで、トリガーワークフローの基本的な使い方が理解できました。次に、より高度なイベントトリガー機能について解説します。

ステップ 1：OSS トリガーの作成

1. オペレーションセンター > スケジューリング設定 > トリガーマネジメント に移動します。

2. 新規トリガー をクリックし、以下のように設定します：

   説明

   各パラメーターの詳細については、「OSS トリガー」をご参照ください。

   • トリガー名：任意の名前（例：oss_new_file_trigger）を入力します。

   • 適用ワークスペース：ワークフローが配置されている対象のワークスペースを選択します。

   • トリガーイベントタイプ： Object Storage Service (OSS) を選択します。

   • トリガーイベント： oss:ObjectCreated:PutObject（またはその他のアップロードイベント）を選択します。

   • バケット名：ご利用の OSS バケットを選択します。

   • ファイル名：監視対象のファイルパスおよび形式を指定します。ワイルドカードが使用可能です。たとえば、input/ ディレクトリ内のすべての .csv ファイルを監視する場合は、input/*.csv を入力します。

   • ロール設定：初回利用時は ワンクリック承認 を実行し、生成されたロール DataWorks-EventBridge-OSS-MNS-Role-************* を選択します。

     ************* は、一意性を保証するためのランダム生成された 13 桁の ID 番号を表します。

3. 確認 をクリックしてトリガーを作成します。

ステップ 2：ワークフローの作成および関連付け

1. クイックスタート：手動トリガーワークフローの作成 の手順に従い、トリガーワークフロー を作成し、名前を process_oss_file_workflow とします。

2. ワークフローキャンバスの右側パネルで、スケジュール設定 > スケジューリングポリシー を選択します。

3. トリガー のドロップダウンリストから、先ほど作成した oss_new_file_trigger を選択します。

   

ステップ 3：イベントパラメーターの解析

1. ツールバーの + ノードを追加 をクリックしてノード一覧を開き、ノードタイプ一覧から Shell ノードをキャンバスにドラッグ＆ドロップして名前を入力します。

2. ノードをダブルクリックし、トリガーイベントからファイルパスを取得・出力するコードを記述します。

   # トリガーがワークフローを起動すると、イベント情報は組み込み変数 workflow.triggerMessage を通じて渡されます。
   # アップロードされたファイルの完全なパスは、${workflow.triggerMessage.data.oss.object.key} を使用して取得できます。
   
   echo "========= OSS ファイル処理を開始 ========="
   message="${workflow.triggerMessage}"
   echo "生の値: ${message}"
   
   # イベントメッセージからファイル名を抽出
   FILE_PATH="${workflow.triggerMessage.data.oss.object.key}"
   echo "新しいファイルが到着しました: ${FILE_PATH}"
   
   # ここに具体的な処理ロジックを追加してください
   
   echo "========= OSS ファイル処理を完了 ========="

   説明

   ${workflow.triggerMessage}：JSON 形式で完全なイベントメッセージ本文を取得します。OSS のメッセージ形式の詳細については、EventBridge の イベントバス > DATAWORKS_TRIGGER_FOR_BUCKET_<OSS_Bucket_Name> > イベントトレーシング > イベント詳細 をご参照ください。

   OSS のメッセージ形式の例

   {
       "datacontenttype": "application/json;charset=utf-8",
       "aliyunaccountid": "1***********9",
       "data": {
           "eventVersion": "1.0",
           "responseElements": {
               "requestId": "69B1***********C0A8"
           },
           "eventSource": "acs:oss",
           "eventTime": "2026-03-11T05:40:45.000Z",
           "requestParameters": {
               "sourceIPAddress": "***********"
           },
           "eventName": "ObjectCreated:PostObject",
           "userIdentity": {
               "principalId": "1***********9"
           },
           "region": "cn-hangzhou",
           "oss": {
               "bucket": {
                   "name": "******",
                   "arn": "acs:oss:cn-hangzhou:1***********9:******",
                   "virtualBucket": "",
                   "ownerIdentity": "1***********9"
               },
               "ossSchemaVersion": "1.0",
               "object": {
                   "size": 59537,
                   "objectMeta": {
                       "mimeType": "text/csv"
                   },
                   "deltaSize": 0,
                   "eTag": "63***********D32",
                   "key": "input/***********.csv"
               }
           }
       },
       "subject": "acs:oss:cn-hangzhou:1***********9:dwoss1024/input/******.csv",
       "aliyunoriginalaccountid": "1***********9",
       "source": "acs.oss",
       "type": "oss:ObjectCreated:PostObject",
       "aliyunpublishtime": "2026-03-11T05:40:45.682Z",
       "specversion": "1.0",
       "aliyuneventbusname": "DATAWORKS_TRIGGER_FOR_BUCKET_******",
       "id": "69B1***********0A8",
       "time": "2026-03-11T05:40:45.000Z",
       "aliyunregionid": "cn-hangzhou"
   }

ステップ 4：デバッグおよび公開

1. デバッグ：

   • ワークフローキャンバスに戻り、実行 ボタンをクリックします。

   • トリガーメッセージ本文 入力欄に、サンプルの OSS イベント（JSON 形式）を貼り付けます。トリガー設定ページからメッセージ形式の例をコピーし、key の値を修正してください。以下は簡易な例です。

     {
       "data": {
         "oss":{
           "object": {
             "key": "input/test_file_20260310.csv" 
           }
         } 
       }
     }

   • 実行 をクリックし、ログに input/test_file_20260310.csv が正しく出力されるか確認します。

2. 公開：デバッグが成功したら、公開 ボタンをクリックして、ワークフローを本番環境に公開します。イベントトリガーは本番環境でのみ有効です。

ステップ 5：本番環境での検証

1. OSS コンソールまたはクライアントを使用して、トリガーで設定したバケットおよびパス（例：input/ ディレクトリ）に CSV ファイルをアップロードします。

   イベントトリガーの検証

   https://eventbridge.console.alibabacloud.com/<regionId>/event-bus/DATAWORKS_TRIGGER_FOR_BUCKET_<OssBucketName>/event-tracing にアクセスし、最近トリガーされたイベントの一覧を確認します。また、イベント詳細 をクリックして、特定のトリガーされたメッセージ（workflow.triggerMessage）を確認することもできます。

   

2. DataWorks の オペレーションセンター > ワンタイムタスク運用管理 > ワンタイムタスク > トリガーワークフロー に移動します。正常に公開された process_oss_file_workflow が表示されます。

   

3. しばらく待った後、DataWorks の オペレーションセンター > ワンタイムタスク O&M > トリガーワークフローインスタンス に移動します。新しいワークフローインスタンスが自動的に起動されます。これをクリックしてログを確認し、ファイルパスが正しく処理されたことを確認します。

シナリオ 2：Kafka メッセージの処理

目的：Kafka トピックをユーザー行動ログ向けにモニターし、新しいメッセージが到着した際に、その内容に応じて異なるロジックを実行するワークフローを起動します。

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

トリガーワークフローのオーケストレーションのコアプロセスは、定期実行ワークフローと同様です。詳細については、「ノードおよびワークフローのオーケストレーション」をご参照ください。

スケジューリングパラメーター

ワークフローキャンバス右側の スケジュール設定 パネルで、ワークフロー 全体のグローバルパラメーターを設定できます。ワークフロー内のすべてのノードでこれらのパラメーターを参照できます。

• 参照構文：ノードのコード内で、ワークフローパラメーターは ${workflow.parameter_name} の形式で参照します。

• パラメーター優先順位：DataWorks のパラメーターには階層的なオーバーライド関係があります。優先順位は次のとおりです：ノードパラメーター > ワークフローパラメーター。

  パラメーターの詳細については、「パラメーター設計およびフロー」をご参照ください。

スケジューリングポリシー

複数のワークフローまたはタスクが同時にトリガーされ、システムリソースのボトルネックが発生した場合、優先度 および 優先度重み付けポリシー を使用して、インテリジェントなリソーススケジューリングを実現できます。これにより、最も重要なタスクが優先的に実行されます。

• コア業務の継続性確保：コア業務ワークフローに高い優先度を設定することで、他の非コアワークフローよりも常に先に実行されるようにします。

• クリティカルパスの所要時間短縮：単一のワークフローインスタンス内で、優先度重み付けポリシー を使用してノードの実行順序に影響を与えることができます。たとえば、下向き重み付け ポリシーでは、クリティカルパス上の依存関係の多いノードに動的に高い重みが割り当てられます。これにより、それらのノードの実行が優先され、ワークフロー全体の実行時間を短縮できます。

  バージョン	バージョン
  優先度	スケジューリングキュー内におけるワークフローインスタンスの絶対的優先度レベルを定義します。利用可能なレベルは 1、3、5、7、8 であり、数値が大きいほど優先度が高くなります。高優先度のタスクまたはワークフローは、低優先度のものよりも常に先にスケジューリングリソースを獲得します。
  優先度重み付けポリシー	同一優先度レベル内で、内部ノード（タスク）の重みをどのように動的に算出するかを定義します。重みが高いノードが実行を優先されます。 重み付けなし：すべてのノードが固定のベースライン重みを持ちます。 下向き重み付け：ノードの重みは動的に調整されます。ノードが持つ上流依存関係が多いほど、その重みは高くなります。この戦略は、DAG（有向非巡回グラフ）におけるクリティカルパス上のノードの実行を優先するのに役立ちます。重みは以下の式で算出されます：初期重み + すべての上流ノードの優先度の合計。
  最大並列インスタンス数	このワークフローが同時に実行できるインスタンスの最大数を制御します。これは同時実行制御およびリソース保護のために使用されます。実行中のインスタンス数が上限に達すると、新たにトリガーされたインスタンスは待機状態になります。この値を 無制限 に設定するか、または最大 100,000 までのカスタム上限値を指定できます。 説明 指定された上限がリソースグループの最大容量を超えた場合、リソースグループの物理的な制限が実際の同時実行ボトルネックとなります。

DataWorks の優先度システムは、階層的なオーバーライドルールに従います：ランタイム仕様 > ノードレベル設定 > ワークフローレベル設定。

1. ワークフローレベル設定（ベースライン）：ワークフローの スケジューリングポリシー で設定され、すべてのノードのデフォルト設定として機能します。

2. ノードレベル設定（ローカル）：ワークフロー内の個々のノードの スケジュール設定 > スケジューリングポリシー で、特定のノードに対してより高い 優先度 を設定できます。これはワークフローレベル設定をオーバーライドします。

3. ランタイム仕様（一時的）： オペレーションセンター で手動実行する際に、ランタイムで優先度をリセット スイッチを使用して設定を指定できます。この設定は最優先され、現在の実行にのみ適用され、永続的な設定を変更しません。