HTTP トリガーノードを使用したテナント間のタスク実行 - DataWorks

DataWorks は、HTTP トリガーノードを提供しており、異なるテナント間でタスクをトリガーして実行できます。このトピックでは、DataWorks で HTTP トリガーノードを使用してテナント間のタスクのトリガーを実装する方法をステップバイステップで説明します。

背景情報

現代の企業では、同じテナント内の複数のテナントまたはリージョンにまたがってデータ処理と分析を実行する必要があることがよくあります。DataWorks は、次のシナリオのために HTTP トリガーノードを提供します：

テナント間でタスクの依存関係と連携が必要なシナリオ。
同じテナント内の異なるスケジューリングシステム間に依存関係が存在するシナリオ (例：システム B は、処理を進める前にシステム A のノードが処理を完了することに依存している)。

テナント間シナリオの図の例：

図の説明：HTTP トリガーノードで構成されたワークフローは、テナント B のオペレーションセンターにデプロイされます。ワークフローがスケジュールされた実行時間に達すると、HTTP トリガーノードより前の上流ノードは、構成されたスケジュールに従って順次実行されます。実行が HTTP トリガーノードに達すると、そのノードとすべての下流タスクは待機状態になります。HTTP トリガーノードがテナント A からトリガーコマンドを受信すると、検証を実行します。検証が成功すると、HTTP トリガーノードとその下流タスクは実行に進みます。

前提条件

2 つのテナントアカウント (つまり、2 つの Alibaba Cloud アカウント) が利用可能であること。
両方のテナントでワークスペースを作成していること。
サーバーレスリソースグループが各テナントの各ワークスペースに関連付けられていること。
トリガーを送信するワークスペースのサーバーレスリソースグループの VPC に、インターネット NAT ゲートウェイと EIP が構成されていること。
重要
Shell ノードが HTTP トリガーノードをトリガーする場合、トリガーコマンドはパブリックネットワーク経由で送信する必要があります。
Shell ノードが存在するワークスペースに MaxCompute コンピューティングリソースが関連付けられていること。詳細については、「コンピューティングリソース管理」をご参照ください。
説明
トリガーとなる Shell ノードによって実行される JAR ファイルは、MaxCompute にアップロードする必要があります。

注意事項

HTTP トリガーノードはトリガーにのみ使用され、開発ロジックは必要ありません。
ご利用のワークスペースの種類に基づいて、対応する操作を実行してください：
- DataStudio のパブリックプレビューに参加しているワークスペースの場合：DataStudio (新バージョン) のトリガー例をご参照ください。
- DataStudio のパブリックプレビューに参加していないワークスペースの場合：DataStudio (旧バージョン) のトリガー例をご参照ください。
  説明
  DataStudio のパブリックプレビューに参加していないワークスペースの場合、HTTP トリガーノード機能は DataWorks Enterprise Edition でのみサポートされます。

DataStudio (新バージョン) のトリガー例

新しい DataStudio を使用したスケジューリングシナリオで、テナント A からテナント B へのノード実行をトリガーする必要がある場合は、以下の手順に従ってください。

テナント B：HTTP トリガーノードを含むワークフローの作成

テナント B のワークスペースで、リモートでトリガーされるワークフローを作成します。このワークフローには、トリガーコマンドを受信する上流の HTTP トリガーノードと、テナント間のトリガー効果を検証するためのビジネスノード (この例では Shell ノード) を含める必要があります。

DataStudio に移動します。
DataWorks ワークスペースリストページに移動し、目的のリージョンに切り替えて目的のワークスペースを見つけ、[操作] 列の [ショートカット] > [Data Studio] をクリックします。
新しいワークフローを作成します。
1. 左側のナビゲーションウィンドウで、アイコンをクリックします。
2. アイコンをクリックして [ワークフローの作成] を選択します。ダイアログボックスで、名前 (例：HTTP_Workflow) を入力し、[OK] をクリックします。
新しいノードを作成します。
1. ワークフロー構成ページで、[全般] > [HTTP トリガー] と [全般] > [Shell] をキャンバスにドラッグします。
2. Shell ノードを HTTP トリガーノードの下流ノードとしてリンクします。
Shell ノードを編集します。
1. Shell ノードにカーソルを合わせ、[ノードを開く] をクリックします。
2. エディターに次のように入力します：
```
echo "DataWorks";
```
3. 右側の [スケジューリング] をクリックし、[リソースグループ] をワークスペースに関連付けられているサーバーレスリソースグループに設定します。
4. 上部のツールバーで [保存] をクリックします。
ワークフローをデプロイします。
1. ワークスペースディレクトリでワークフロー HTTP_Workflow を見つけ、右側の [スケジューリング] で [インスタンス生成モード] を [デプロイ後すぐに生成] に設定します。
2. ワークフロー編集ページの上部ツールバーで [保存] をクリックします。
3. [変更レビュー] ダイアログボックスで、[保存] をクリックします。
4. ツールバーのアイコンをクリックし、[本番環境へのデプロイを開始] をクリックします。詳細については、「ノードとワークフローのデプロイメント」をご参照ください。
HTTP インスタンスのパラメーターを記録します。
HTTP トリガーノードはすぐに自動トリガーインスタンスを生成するため、[オペレーションセンター] に移動して HTTP インスタンスのパラメーター情報を表示および記録できます。
1. オペレーションセンターページに移動します。
  DataWorks コンソールにログインします。上部のナビゲーションバーで、目的のリージョンを選択します。左側のナビゲーションウィンドウで、[データ開発と O&M] > [オペレーションセンター] を選択します。表示されたページで、ドロップダウンリストから目的のワークスペースを選択し、[オペレーションセンターへ移動] をクリックします。
2. 左側のナビゲーションウィンドウで、[自動トリガーノード O&M] > [自動トリガーインスタンス] をクリックします。
3. HTTP トリガーノードインスタンスを見つけ、その [タスク ID] と [スケジュール時刻] を記録します。
  説明
  HTTP トリガーノードインスタンスの名前にカーソルを合わせると、その [タスク ID] を表示できます。

ローカル環境：サンプルコードの準備

pom 依存関係を追加します。

TriggerSchedulerTaskInstance デバッグページに移動し、[SDK サンプルコード] タブで完全な [SDK インストール情報] を表示できます。

<build>
       <plugins>
           <plugin>
               <groupId>org.apache.maven.plugins</groupId>
               <artifactId>maven-assembly-plugin</artifactId>
               <version>3.3.0</version>
               <configuration>
                   <archive>
                       <manifest>
                           <mainClass>com.example.demo.CrossTenantTriggerSchedulerTaskInstance</mainClass>  <!-- メインクラスに置き換えます -->
                       </manifest>
                   </archive>
                   <descriptorRefs>
                       <descriptorRef>jar-with-dependencies</descriptorRef>
                   </descriptorRefs>
               </configuration>
               <executions>
                   <execution>
                       <id>make-assembly</id>
                       <phase>package</phase> <!-- パッケージングフェーズで実行 -->
                       <goals>
                           <goal>single</goal>
                       </goals>
                   </execution>
               </executions>
           </plugin>
       </plugins>
   </build>
   <dependencies>
       <dependency>
           <groupId>com.aliyun</groupId>
           <artifactId>dataworks_public20240518</artifactId>
           <version>6.2.0</version>
       </dependency>
   </dependencies>

重要

コード開発が完了したら、mainClass パラメーターを作成した Java メインクラスの名前に置き換えます。メインクラス名のフォーマットは、完全なパッケージ名とクラス名です。例：com.example.demo.CrossTenantTriggerSchedulerTaskInstance。

コード開発。

package com.example.demo;
   import com.aliyun.dataworks_public20240518.Client;
   import com.aliyun.dataworks_public20240518.models.TriggerSchedulerTaskInstanceRequest;
   import com.aliyun.dataworks_public20240518.models.TriggerSchedulerTaskInstanceResponse;
   import com.aliyun.teautil.models.RuntimeOptions;

   import java.text.SimpleDateFormat;
   import java.util.Calendar;

   public class CrossTenantTriggerSchedulerTaskInstance {
       // Alibaba Cloud DataWorks クライアントを作成
       public static Client createClient20240518(String accessId, String accessKey, String endpoint) throws Exception {
           // OpenAPI 設定オブジェクトを初期化
           com.aliyun.teaopenapi.models.Config config = new com.aliyun.teaopenapi.models.Config();
           config.setAccessKeyId(accessId); // AccessKey ID を設定
           config.setAccessKeySecret(accessKey); // AccessKey Secret を設定
           config.setEndpoint(endpoint); // エンドポイントを設定
           return new Client(config); // 初期化されたクライアントインスタンスを返す
       }
       
       // DataWorks ノードの実行をトリガー
       public static TriggerSchedulerTaskInstanceResponse runTriggerScheduler(Client client, Long nodeId, String EnvType, Long TriggerTime) throws Exception {
           TriggerSchedulerTaskInstanceRequest request = new TriggerSchedulerTaskInstanceRequest(); // API リクエストオブジェクトを作成
           request.setTaskId(nodeId); // ノード ID を設定
           request.setEnvType(EnvType); // 環境を設定
           request.setTriggerTime(TriggerTime); // スケジュールされたトリガー時刻 (ミリ秒単位のタイムスタンプ) を設定
           RuntimeOptions runtime = new RuntimeOptions(); // ランタイム設定を初期化
           return client.triggerSchedulerTaskInstanceWithOptions(request, runtime); // API 操作を実行して応答を返す
       }
       
       // メインメソッド
       public static void main(String[] args) throws Exception {
           // ノード ID を初期化 (値の例)
           String nodeId1 = "";
           // 環境を初期化 (値の例)
           String EnvTypeStr = "";
           // スケジュール時刻を初期化 (値の例)
           String cycTimeStr = "";
           
           // CLI 引数を解析：nodeId、環境タイプ、スケジュール時刻
           int i;
           for(i = 0; i < args.length; ++i) {
               if (i == 0) {
                   nodeId1 = args[i];
               } else if (i == 1) {
                   EnvTypeStr = args[i];
               } else if (i == 2) {
                   cycTimeStr = args[i];
               }
           }
           
           // 文字列を Long 型のノード ID に変換
           Long nodeId = Long.parseLong(nodeId1);
           
           // 使用方法の説明を出力
           System.out.println("Usage: java -jar test-1.0-SNAPSHOT.jar nodeId EnvTypeStr cycTimeParam");
           
           // スケジュール時刻を解析してタイムスタンプに変換
           SimpleDateFormat sdft = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
           Calendar calendar = Calendar.getInstance();
           calendar.setTime(sdft.parse(cycTimeStr)); // 時刻文字列を解析
           Long cycTime = calendar.getTimeInMillis(); // ミリ秒単位のタイムスタンプを取得
           
           // デバッグ情報 (タイムスタンプ) を出力
           System.out.println("Scheduled timestamp: " + cycTime);
           
           // Alibaba Cloud サービスパラメーターを設定
           String endpoint = "dataworks.cn-hangzhou.aliyuncs.com"; // サービスエンドポイント
           String accessId = "xxx"; // アクセスキー ID
           String accessKey = "xxx"; // アクセスキーシークレット
           
           // Alibaba Cloud クライアントインスタンスを作成
           Client client = createClient20240518(accessId, accessKey, endpoint);
           
           // トリガーノード操作を実行
           TriggerSchedulerTaskInstanceResponse response = runTriggerScheduler(client, nodeId, EnvTypeStr, cycTime);
           
           // API 応答結果 (JSON フォーマット) を出力
           System.out.println(com.aliyun.teautil.Common.toJSONString(com.aliyun.teautil.Common.toMap(response)));
       }
   }

上記のコードの endpoint、accessId、accessKey パラメーターを、次のパラメーターの説明に従って、ビジネスで必要な実際の値に置き換えます。

パラメーター	説明
endpoint	対象の HTTP トリガーノードが配置されているワークスペースのサービスアドレス。詳細については、「Alibaba Cloud API エンドポイント」をご参照ください。
accessId	対象の HTTP トリガーノードが配置されている Alibaba Cloud アカウントの `AccessKey ID`。	DataWorks コンソールにログインし、右上隅のプロファイル画像にカーソルを合わせ、[AccessKey] に移動して、対象の HTTP トリガーノードが配置されている RAM ユーザーの `AccessKey ID` と `AccessKey Secret` を取得できます。警告 Alibaba Cloud アカウントの `AccessKey` は完全な権限を持っています。漏洩すると、深刻なセキュリティリスクをもたらします。対象のワークスペースに対して [ワークスペース管理者] ロールのみを持つ RAM ユーザーの AccessKey を使用することを推奨します。
accessKey	対象の HTTP トリガーノードが配置されている Alibaba Cloud アカウントの `AccessKey Secret`。

上記のコードをパッケージ化して、サフィックスが jar-with-dependencies.jar の JAR パッケージを生成します。

テナント A：Shell ノードの設定

テナント A のワークスペースで Shell ノードを設定し、テナント B のワークスペースの HTTP トリガーノードをトリガーするには、次の手順に従います。

Data Studio に移動します。
DataWorks のワークスペースリストページに移動し、対象のリージョンに切り替え、対象のワークスペースを見つけます。[操作] 列で、[ショートカット] > [Data Studio] をクリックします。
JAR ファイルをアップロードします。
1. 左側のナビゲーションウィンドウで、アイコンをクリックします。
2. [リソース管理] ページで、[作成] ボタンまたはアイコンをクリックし、[リソースの作成] > [MaxCompute Jar] を選択します。
3. [リソースまたは関数の作成] ダイアログボックスで、リソース名 http_node_work.jar を入力し、[OK] をクリックします。
4. アップロードリソースの詳細ページで、[アップロード] をクリックして、「ローカル環境：サンプルコードの準備」ステップで生成した JAR パッケージをアップロードし、[データソース] にバインドした MaxCompute データソースを選択します。
リソースを保存して公開します。
準備した JAR リソースがアップロードされた後、ツールバーのボタンをクリックしてデプロイメントパネルを表示し、[本番環境へのデプロイを開始] をクリックして、デプロイメントチェックプロセスの操作を順に完了します。詳細については、「ノードとワークフローのデプロイメント」をご参照ください。
トリガー側の Shell ノードを作成します。
1. 左側のナビゲーションウィンドウで、をクリックし、[ワークスペースディレクトリ] の横にあるアイコンをクリックします。
2. [全般] > [Shell] を選択すると、[ノードの作成] ダイアログボックスが表示されます。
3. ノードの [名前] をカスタマイズし、[OK] をクリックしてノード編集ページに移動します。

トリガー側の Shell ノードを編集します。

Shell ノード編集ページの左側のナビゲーションウィンドウでアイコンをクリックして、アップロードした JAR リソース (http_node_work.jar) を見つけます。
アップロードした JAR リソースを右クリックし、[リソースの参照] を選択します。
Shell ノードでトリガーコード実行パラメーター情報を入力します。

##@resource_reference{"http_node_work.jar"}
 java -jar http_node_work.jar nodeId "EnvTypeStr" "cycTimeParam"

パラメーター	説明
java -jar	JAR 実行コマンド。
http_node_work.jar	参照したリソースの名前。
nodeId	「テナント B：HTTP トリガーノードを含むワークフローの作成」ステップで記録した HTTP トリガーノードの [タスク ID]。
EnvTypeStr	対象の HTTP トリガーノードのプロジェクト環境。このトピックでは、本番環境に公開された HTTP トリガーノードを実行するため、パラメーターを `Prod` に設定します。開発環境で HTTP トリガーノードを実行する必要がある場合は、パラメーターを `Dev` に設定します。
cycTimeParam	「テナント B：HTTP トリガーノードを含むワークフローの作成」ステップで記録した HTTP トリガーノードタスクの [スケジュール時刻]。時刻のフォーマットは `yyyy-MM-dd HH:mm:ss` です。

トリガー側の Shell ノードを設定します。
Shell ノード編集ページの右側にある [スケジューリング] で、[スケジュールポリシー] > [リソースグループ] を見つけます。ワークスペースにバインドされているサーバーレスリソースグループを [リソースグループ] として選択します。

実行と結果の表示

テナント A の下で Shell ノードを実行して、テナント B の HTTP トリガーノードとその後のノードタスクの実行をトリガーします。

トリガー側の Shell ノードを実行します。
テナント A の「Shell ノードの設定」ステップで設定した Shell ノードの上部ツールバーにある [実行時間] をクリックします。
実行結果を表示します。
テナント B に移動し、次の手順に従って実行結果を表示する必要があります。
1. オペレーションセンターページに移動します。
  DataWorks コンソールにログインします。上部のナビゲーションバーで、目的のリージョンを選択します。左側のナビゲーションウィンドウで、[データ開発と O&M] > [オペレーションセンター] を選択します。表示されたページで、ドロップダウンリストから目的のワークスペースを選択し、[オペレーションセンターへ移動] をクリックします。
2. 左側のナビゲーションウィンドウで、[自動トリガーノード O&M] > [自動トリガーインスタンス] をクリックします。
3. トリガーしたい HTTP トリガーノードインスタンスを見つけ、実行結果を確認します。

DataStudio (旧バージョン) のトリガー例

旧バージョンの DataStudio で、テナント A からテナント B へのノード実行をスケジューリングシナリオでトリガーする必要がある場合は、以下の内容をご参照ください。

テナント B：HTTP トリガーノードワークフローの作成

テナント B のワークスペースに、リモートでトリガーする必要があるワークフローを作成します。このワークフローには、トリガーを待つビジネスノード (このトピックでは Shell ノードを例として使用) と、トリガーコマンドを受信する上流の HTTP トリガーノードを含め、後でテナント間のトリガー効果を検証します。

HTTP トリガーノードを作成します。詳細については、「HTTP トリガーノードの作成」をご参照ください。
Shell ノードを作成します。詳細については、「Shell ノードの作成」をご参照ください。
ノード編集ページに次の内容を入力し、Shell ノード編集ページの右側にある [スケジューリング] で [再実行をサポート] と [リソースグループ] を設定します。ノード構成情報を [保存] します。
```
echo "DataWorks";
```
ノードをデプロイしてオペレーションセンターに公開します。
HTTP トリガーノードインスタンスを見つけ、その [タスク ID]、[スケジュール時刻]、[データタイムスタンプ] を記録します。
HTTP トリガーノードはデプロイメントの 1 日後に自動トリガーインスタンスを生成するため、翌日ワークフローの右上隅にあるアイコンをクリックし、[オペレーションセンター] の [自動トリガーノード O&M] > [自動トリガーインスタンス] で HTTP インスタンスのパラメーター情報を表示できます。
説明
HTTP トリガーノードインスタンスの名前にカーソルを合わせると、その [タスク ID] を表示できます。