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

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

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

2. 左側のナビゲーションウィンドウで をクリックします。[ビジネスフロー] の横にある > [ワークフローの作成] をクリックして [ワークフローの作成] ページを開きます。

3. [ワークフローの作成] ページで、[スケジュールタイプ] を [トリガー実行] に設定し、ワークフローの [名前] を入力して [OK] をクリックします。

ステップ 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. [OK] をクリックしてトリガーを作成します。

ステップ 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 "========= 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 で [イベントバス] > 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 の値) を表示することもできます。

   

2. DataWorks で、[オペレーションセンター] > [手動トリガーノード O&M] > [手動タスク] > [トリガーワークフロー] に移動します。公開された process_oss_file_workflow がリストに表示されます。

   

3. しばらくすると、新しいワークフローインスタンスが自動的にトリガーされ、実行されます。DataWorks の [オペレーションセンター] > [手動トリガーノード O&M] > [手動インスタンス] で表示できます。インスタンスをクリックしてログを表示し、ファイルパスが正しく処理されたことを確認します。

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

目的：Kafka トピックを監視してユーザーの行動ログを取得します。新しいメッセージが到着したときにワークフローをトリガーし、その内容に基づいて異なるロジックを実行します。

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

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

スケジュールパラメーター

ワークフローキャンバスの右側にある [プロパティ] パネルで、ワークフローのグローバルパラメーターを設定できます。その中のすべてのノードはこれらのパラメーターを参照できます。

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

• パラメーターの優先度：DataWorks のパラメーターには階層的な上書き関係があります。優先順位は、ノードパラメーター > ワークフローパラメーターです。

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

スケジュールポリシー

複数のワークフローが同時にトリガーされ、リソースのボトルネックが発生した場合、[優先度] と [優先度重み付けポリシー] を使用してリソーススケジューリングを管理し、最も重要なタスクが最初に実行されるようにします。

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

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

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

DataWorks の優先度システムは、階層的な上書きルールに従います：ランタイム指定 > ノードレベルの構成 > ワークフローレベルの構成。

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

2. ノードレベルの構成 (上書き)：特定のノードの [プロパティ] > [スケジュールポリシー] パネルでより高い [優先度] を設定することで、ワークフローレベルの設定を上書きできます。

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