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. デプロイが成功したら、オペレーションセンター > ワンタイムタスクの運用保守 > ワンタイムタスク > トリガー式ワークフローに移動します。

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 トリガーを選択します。

   トリガーを選択すると、内部タスクで${workflow.triggerMessage}を使用してメッセージ本文全体を取得するか、${workflow.triggerMessage.xxx}を使用してメッセージ本文内の特定のフィールドの値を取得できます。

ステップ 3：ノードの開発とイベントパラメーターの解析

1. ツールバーの+ ノードの追加をクリックしてノードリストを開きます。左側のノードタイプリストからShellノードをキャンバス上にドラッグし、名前を入力して作成を完了します。

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

   # トリガーがワークフローを起動すると、イベント情報は組み込み変数 workflow.triggerMessage を介して渡されます
   # アップロードされたファイルの完全なパスは ${workflow.triggerMessage.data.oss.object.key} で取得できます
   echo "========= Start Processing OSS File ========="
   message="${workflow.triggerMessage}"
   echo "Raw Value: ${message}"
   # イベントメッセージからファイル名を抽出
   FILE_PATH="${workflow.triggerMessage.data.oss.object.key}"
   echo "A new file has arrived: ${FILE_PATH}"
   # ここに具体的な処理ロジックを追加
   echo "========= Finish Processing OSS File ========="

   説明

   ${workflow.triggerMessage}：JSON 形式で完全なイベントメッセージ本文を取得します。OSS の具体的なメッセージフォーマットは、EventBridge > EventBridge > DATAWORKS_TRIGGER_FOR_BUCKET_<OSS_Bucket_Name> > イベントトレーシング > イベント詳細から取得できます。

   OSS メッセージフォーマットのサンプル（workflow.triggerMessage のフォーマット）を表示します。

   {
       "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）を確認することもできます。

   イベント一覧には、イベントソースがObject Storage Service (OSS) / acs.oss、イベントタイプがoss:ObjectCreated:PostObjectのトリガー記録が表示されます。

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

3. しばらく待ってから、DataWorks オペレーションセンター > ワンタイムタスクの運用保守 > トリガー式ワークフローのインスタンスに移動します。新しいワークフローインスタンスが自動的にトリガーされています。クリックしてログを確認し、ファイルパスが正しく処理されていることを検証します。

   ===== Start Processing 0SS File =====
   Raw Value: {"datacontenttype":"application/json;charset=utf-8","aliyunaccountid":"1162423445433459","data":{"eventVersion":"1.0","responseElements":{"requestId":"69B1083F7A439F343040ABCD"},
   "eventSource":"acs:oss","eventTime":"2026-03-11T06:14:23.000Z","requestParameters":{"sourceIPAddress":"140.205.11.13"},"eventName":"0bjectCreated:Post0bject","userIdentity":{"principalId":"1162423445433459"},
   "region":"cn-hangzhou","oss":{"bucket":{"name":"dwoss1024","arn":"acs:oss:cn-hangzhou:1162423445433459:dwoss1024","virtualBucket":"","ownerIdentity":"1162423445433459"},"ossSchemaVersion":"1.0","object":{"size":59537,"objectMeta":
   {"mimeType":"text/csv"},"deltaSize":0,"eTag":"63B4BA5A45AEFC679B9A917E8DDF0D32","key":"input/2013-2020-global-PS4-game-sales.csv"}}},"subject":"acs:oss:cn-hangzhou:1162423445433459:dwoss1024/input/2013-2020-global-PS4-game-sales.csv","aliyunoriginalaccountid":"1162423445433459","source":"acs.oss","type":"oss:0bjectCreated:Post0bject","aliyunpublishtime":"2026-03-11T06:14:23.959Z","specversion":"1.0","aliyuneventbusname":"DATAW0RKS_TRIGGER_F0R_BUCKET_dwoss1024","id":"69B1083F7A439F343040ABCD","time":"2026-03-11T06:14:23.000Z","aliyunregionid":"cn-hangzhou"}A new file has arrived: input/2013-2020-global-PS4-game-sales.csv
   ====== Finish Processing 0SS File =========

シナリオ 2：Kafka メッセージ到着によるリアルタイム計算の駆動

目的：Kafka のユーザービヘイビアログを監視し、新しいメッセージが到着した際にワークフローをトリガーして解析し、内容に基づいて異なるロジックを実行します。

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

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

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

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

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

• パラメーター優先度：DataWorks のパラメーターは階層的なオーバーライドルールに従います。優先度の順序は次のとおりです：ノードパラメーター > ワークフローパラメーター。

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

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

複数のワークフローやタスクが同時にトリガーされ、システムリソースがボトルネックになる場合、優先度および重み付け戦略を使用してインテリジェントなリソーススケジューリングを実現し、最も重要なタスクが最初に実行されるようにします。

• コアビジネスの保護：コアビジネスワークフローに高い優先度を設定することで、他の非コアワークフローよりも常に先に実行されるようにします。

• クリティカルパスの持続時間短縮：同一ワークフローインスタンス内では、優先度加重ポリシーを使用してノードの実行順序に影響を与えます。たとえば、ダウンストリーム重み付け戦略では、より多くの上流依存関係を持つノードがより高い動的重みを受け取り、先に実行されることで、全体のワークフロー実行時間を効果的に短縮できます。

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

DataWorks の優先度システムは階層的なオーバーライドルールに従います：実行時指定 > ノードレベル構成 > ワークフローレベル構成。

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

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

3. 実行時指定（一時的）：オペレーションセンターで手動トリガー実行時にランタイムリセットの優先度トグルを通じて指定します。この構成は最高優先度を持ち、現在の実行に対してのみ有効で、永続的な構成は変更しません。