:ベストプラクティス：サードパーティのスケジューリングシステムの統合

依存する DataWorks ノードをサブスクライブした後、DataWorks インスタンスのステータスに基づいてタスクをトリガーするようにサードパーティのスケジューリングシステムを構成する必要があります。システムが依存する DataWorks ノードの実行成功通知を受信すると、サードパーティシステム内のタスクが開始されます。以下のサンプルコードに構成例を示します。

package com.aliyun.dataworks.demo;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.aliyun.dataworks.services.LocalScheduleEventService;
import com.aliyun.dataworks.utils.Constants;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @author dataworks demo
 */
@RestController
@RequestMapping("/event")
public class EventController {
  @Autowired
  private LocalScheduleEventService localScheduleEventService;

  /**
   * EventBridge からプッシュされたメッセージを受信します。
   *
   * @param jsonParam
   */
  @PostMapping("/consumer")
  public void consumerEventBridge(@RequestBody String jsonParam) {
    JSONObject jsonObj = JSON.parseObject(jsonParam);
    String eventCode = jsonObj.getString(Constants.EVENT_CODE_FILED);
    if (Constants.INSTANCE_STATUS_EVENT_CODE.equals(eventCode)) {
      JSONObject dataParam = JSON.parseObject(jsonObj.getString("data"));
      // スケジュールされたタスクインスタンスが待機を開始した特定の時刻。
      System.out.println("beginWaitTimeTime: " + dataParam.getString("beginWaitTimeTime"));
      // DagId
      System.out.println("dagId: " + dataParam.getString("dagId"));
      // DAG のタイプ。有効な値：
      // 0：定期的にスケジュールされたタスク
      // 1：ワンタイムタスク
      // 2：スモークテスト
      // 3：データバックフィル
      // 4：手動トリガーワークフロー
      // 5：一時的なワークフロー
      System.out.println("dagType: " + dataParam.getString("dagType"));
      // タスクインスタンスのスケジューリングタイプ。有効な値：
      // NORMAL(0)：通常スケジュールされたタスク。このタスクは毎日スケジュールされます。
      // MANUAL(1)：ワンタイムタスク。このタスクは毎日スケジュールされません。
      // PAUSE(2)：凍結されたタスク。このタスクは毎日スケジュールされますが、スケジューリング開始時にステータスが直接「失敗」に設定されます。
      // SKIP(3)：ドライランタスク。このタスクは毎日スケジュールされますが、スケジューリング開始時にステータスが直接「成功」に設定されます。
      // SKIP_UNCHOOSE(4)：一時的なワークフローで選択されなかったタスク。このタイプのタスクは一時的なワークフローにのみ存在し、スケジューリング開始時にステータスが直接「成功」に設定されます。
      // SKIP_CYCLE(5)：週次または月次のタスクで、スケジューリングサイクルがまだ到来していないもの。このタスクは毎日スケジュールされますが、スケジューリング開始時にステータスが直接「成功」に設定されます。
      // CONDITION_UNCHOOSE(6)：アップストリームインスタンスに分岐 (IF) ノードがあるが、この子孫ノードが分岐ノードによって選択されなかった場合。タスクは直接ドライランタスクとして設定されます。
      // REALTIME_DEPRECATED(7)：リアルタイムで生成されたが期限切れになった定期インスタンス。このタイプのタスクのステータスは直接「成功」に設定されます。
      System.out.println("taskType: " + dataParam.getString("taskType"));
      // タスクインスタンスが変更された時刻。
      System.out.println("modifyTime: " + dataParam.getString("modifyTime"));
      // タスクインスタンスが作成された時刻。
      System.out.println("createTime: " + dataParam.getString("createTime"));
      // ワークスペースの ID。ListProjects を呼び出してワークスペース ID を表示できます。
      System.out.println("appId: " + dataParam.getString("appId"));
      // スケジュールされたタスクインスタンスが存在するワークスペースのテナント ID。
      System.out.println("tenantId: " + dataParam.getString("tenantId"));
      // スケジュールされたタスクインスタンスの操作コード。このフィールドは無視できます。
      System.out.println("opCode: " + dataParam.getString("opCode"));
      // ワークフローの ID。定期タスクインスタンスのワークフロー ID はデフォルトで 1 です。手動トリガーワークフローまたは内部ワークフロータスクインスタンスの場合、これは実際のワークフロー ID です。
      System.out.println("flowId: " + dataParam.getString("flowId"));
      // スケジュールされたタスクインスタンスに対応するノードの ID。
      System.out.println("nodeId:" + dataParam.getString("nodeId"));
          // スケジュールされたタスクインスタンスがリソースを待機し始めた特定の時刻。
          System.out.println("beginWaitResTime: " + dataParam.getString("beginWaitResTime"));
          // スケジュールされたタスクインスタンス ID。
          System.out.println("taskId: " + dataParam.getString("taskId"));
          // タスクのステータス。有効な値：
          // 0 (未実行)
          // 2 (dueTime または cycleTime で指定されたスケジュール時刻を待機中)
          // 3 (リソースを待機中)
          // 4 (実行中)
          // 7 (データ検証のために Data Quality に送信済み)
          // 8 (分岐条件をチェック中)
          // 5 (失敗)
          // 6 (成功)
          System.out.println("status: " + dataParam.getString("status"));

          // DataWorks からノード完了イベントをサブスクライブした後、ローカルスケジューリングノードをトリガーして実行します。
          localScheduleEventService.triggerLocalNode(dataParam);
      } else {
          System.out.println("他のイベントのフィルタリングに失敗しました。構成手順を確認してください。");
      }
  }
}

以下のサンプルコードは、サードパーティシステムが RunTriggerNode API を呼び出して HTTP トリガーノードをトリガーする方法を示しています。

• HttpTriggerNodeService クラスのサンプル実装。

  package com.aliyun.dataworks.services;
  
  import com.aliyun.dataworks_public20200518.models.RunTriggerNodeRequest;
  import com.aliyun.dataworks_public20200518.models.RunTriggerNodeResponse;
  import org.springframework.beans.factory.annotation.Autowired;
  import org.springframework.stereotype.Service;
  
  /**
   * @author dataworks demo
   */
  @Service
  public class HttpTriggerNodeService {
  
    @Autowired
    private DataWorksOpenApiClient dataWorksOpenApiClient;
  
    /**
     * @return
     */
    public boolean triggerNode(Long appId, Long nodeId, Long bizDate, Long cycleTime) {
      try {
        RunTriggerNodeRequest runTriggerNodeRequest = new RunTriggerNodeRequest();
  
        // NodeId を設定します。これはトリガーノードの ID です。ListNodes API を呼び出すことでノード ID を取得できます。
        runTriggerNodeRequest.setNodeId(nodeId);
  
        // CycleTime を設定します。これはトリガーノードのタスクのランタイムの UNIX タイムスタンプです。HTTP トリガーノードのスケジューリング構成でノードに指定されたランタイムをタイムスタンプに変換する必要があります。
        // HTTP トリガーノードとスケジューリングシステムが異なるタイムゾーンにある場合は、トリガーノードのタイムゾーンに基づいて時刻を構成します。
        // たとえば、HTTP トリガーノードが中国 (北京) リージョンにあり、その Cyctime が 18:00 (UTC+8) であるが、スケジューリングシステムが米国 (西部) リージョンにある場合、18:00 (UTC+8) に対応するタイムスタンプを構成する必要があります。
        runTriggerNodeRequest.setCycleTime(cycleTime);
  
  
        // BizDate を設定します。これはトリガーノードインスタンスのデータタイムスタンプです。データタイムスタンプを UNIX タイムスタンプに変換する必要があります。
        // データタイムスタンプはランタイムの前日を指し、日単位で正確です (時刻部分は 00:00:00)。たとえば、ランタイムが 2020 年 11 月 25 日の場合、ビジネス時刻は 2020112400000000 です。この時刻を UNIX タイムスタンプに変換します。
        // HTTP トリガーノードとスケジューリングシステムが異なるタイムゾーンにある場合は、トリガーノードのタイムゾーンに基づいて時刻を構成します。
        runTriggerNodeRequest.setBizDate(bizDate);
  
        // AppId (appId=projectId) を設定します。これはトリガーノードが属する DataWorks ワークスペースの ID です。GetNode を呼び出すことで、ノードに対応する projectId を見つけることができます。
        runTriggerNodeRequest.setAppId(appId);
  
        RunTriggerNodeResponse runTriggerNodeResponse = dataWorksOpenApiClient.createClient().runTriggerNode(runTriggerNodeRequest);
        System.out.println(runTriggerNodeResponse.getBody().getRequestId());
        return runTriggerNodeResponse.getBody().getSuccess();
      } catch (Exception e) {
        e.printStackTrace();
      }
      return false;
    }
  }
  

• サードパーティのスケジューリング状態機械のサンプル実行。

      @Scheduled(cron = "0 0/30 * * * ? ")
      public void schedule() {
          //TODO：ローカルスケジューリングをシミュレートして、HTTP トリガーノードを介して DataWorks の定期ノードを呼び出します。
          triggerDwScheduleNode();
      }

• triggerDwScheduleNode メソッドのサンプル実装

      /**
       * この例では、日次スケジューリング頻度のトリガーノードを使用して、トリガーノードを呼び出すために必要なパラメーター情報を取得する方法を示します。
       */
      public void triggerDwScheduleNode() {
          Date gmtDate = getTimeByZeroEnd(new Date());
          GregorianCalendar gc = (GregorianCalendar) GregorianCalendar.getInstance();
          gc.setTime(gmtDate);
          gc.add(Calendar.DATE, -1);
          Long bizDate = gc.getTimeInMillis();
          GetNodeResponseBody.GetNodeResponseBodyData node = nodeService.getNode(nodeId);
          if (node != null) {
              String bizDateStr = getTimeInExpress(gc.getTime(), "yyyy-MM-dd HH:mm:ss");
              ListInstancesResponseBody.ListInstancesResponseBodyData instances = nodeService.getInstance(nodeId, node.getProjectId(), bizDateStr);
              if (!CollectionUtils.isEmpty(instances.getInstances())) {
                  ListInstancesResponseBody.ListInstancesResponseBodyDataInstances instance = instances.getInstances().get(0);
                  httpTriggerNodeService.triggerNode(node.getProjectId(), node.getNodeId(), bizDate, instance.getCycTime());
              }
          }
  
      }

プロジェクトのダウンロード：

• 前提条件：Java 8 以降および Maven ビルドツール。

• プロジェクトのダウンロードリンク：schedule-integration-demo.zip。

プロジェクトをダウンロードした後、プロジェクトのルートディレクトリに移動し、次のコマンドを実行します：

mvn clean package -Dmaven.test.skip=true spring-boot:repackage

実行可能な JAR パッケージを取得した後、次のコマンドを実行します：

java -jar target/schedule-integration-demo-1.0.jar

コマンドが実行された後、ブラウザでhttp://localhost:8080/indexと入力します。"hello world!"が返された場合、アプリケーションは正常にデプロイされています。