SchedulerX:ビジュアル MapReduce モデル

一度に複数のアカウントを処理する

例: 一度に複数の銀行口座を処理したいとします。各銀行口座は独立したタスクと見なされ、クラスター内でグローバル並列モードで実行できます。タスクリスト内の各タスクには、簡単にクエリできるようにアカウント情報を表示する必要があります。これにより、各銀行口座の処理ステータスと実行詳細を迅速に理解できます。以下のサンプルコードをご参照ください。

1. 各カスタムアカウント情報サブタスクオブジェクトのラベル情報を表示するには、com.alibaba.schedulerx.worker.processor.BizSubTask インターフェイスとその labelMap メソッドを実装する必要があります。

   コードの表示

   public class ParallelAccountInfo implements BizSubTask {
   
       /**
        * プライマリキー
        */
       private long id;
   
       private String name;
   
       private String accountId;
   
       public ParallelAccountInfo(long id, String name, String accountId) {
           this.id = id;
           this.name = name;
           this.accountId = accountId;
       }
   
       /**
        * タスクにラベルを指定するために使用される labelMap メソッドを実装します。
        * @return
        */
       @Override
       public Map<String, String> labelMap() {
           Map<String, String> labelMap = new HashMap();
           labelMap.put("Account name", name);
           return labelMap;
       }
   }

   各タスクオブジェクトにインターフェイスを実装すると、タスクリストには各タスクオブジェクトのアカウント名などの一意のラベル情報が表示されます。ラベル情報を使用して、各銀行口座のビジネス処理ステータスを理解し、ラベルで銀行口座を検索できます。

   

2. Processor を使用して各銀行口座のビジネスロジックを処理し、com.alibaba.schedulerx.worker.processor.MapReduceJobProcessor を継承します。

   コードの表示

   public class ParallelJob extends MapReduceJobProcessor {
   
       private static final Logger logger = LoggerFactory.getLogger("schedulerx");
   
       @Override
       public ProcessResult reduce(JobContext context) throws Exception {
           return new ProcessResult(true);
       }
   
       @Override
       public ProcessResult process(JobContext context) throws Exception {
           if(isRootTask(context)){
               logger.info("Build parallel computing tasks.");
               List<ParallelAccountInfo> list = new LinkedList();
               /**
                *  ルートタスクの場合、並列計算タスクオブジェクトを構築します。
                *  実際のシナリオでは、必要に応じてタスクオブジェクトをロードし、タスクオブジェクトに BizSubTask インターフェイスを実装できます。
                *  サンプルシナリオ:
                *  1. データベースから未処理の顧客アカウント情報をロードします。
                *  2. 省、市、区などのエリアごとにジョブを分散するためのエリアテーブルを構築します。
                *  3. 家電製品、日用品、食品などのビジネスラベルを使用してタスクを分類します。
                *  4. 時間ベースの値を使用してタスクを分類します。たとえば、1 月や 2 月などの月を使用します。
                */
               for(int i=0; i < 20; i++){
                   list.add(new ParallelAccountInfo(i, "CUS"+StringUtils.leftPad(i+"", 4, "0"),
                           "AC"+StringUtils.leftPad(i+"", 12, "0")));
               }
               return map(list, "transfer");
           }else {
               /**
                * 非ルートタスクの場合、タスクを処理するためのタスク情報を取得します。
                */
               ParallelAccountInfo obj = (ParallelAccountInfo)context.getTask();
               // 取得した obj タスク情報に基づいてタスクを処理します。
               // 何らかの処理
               logger.info("Process task information :{}", JSON.toJSONString(obj));
               return new ProcessResult(true);
           }
       }
   }

   ジョブを開発してデプロイした後、SchedulerX コンソールでスケジュールされたタスクを構成して、指定した時点でジョブを実行します。詳細については、このトピックの「手順」セクションをご参照ください。

ジョブ構成

1. SchedulerX コンソールにログインします。左側のナビゲーションウィンドウで、[タスク管理] をクリックします。

2. [タスク管理] ページで、[タスクの作成] をクリックします。

3. [タスクの作成] パネルで、[実行モード] ドロップダウンリストから [ビジュアル MapReduce] を選択します。

   

4. [詳細設定] セクションで、必要に応じてパラメーターを構成します。その他のパラメーターの詳細については、「ジョブ管理の高度なパラメーター」をご参照ください。

   パラメーター	説明
   配布ポリシー	ポーリングスキーム: システムは、各ワーカーに同数のタスクを均等に分散します。このポリシーは、各ワーカーがタスクの処理にほぼ同じ時間を必要とするシナリオに適しています。これがデフォルト値です。 負荷最適ポリシー: マスターノードはワーカーノードの負荷を自動的に検出します。このポリシーは、サブタスクの処理時間が大幅に異なるシナリオに適しています。 説明 このパラメーターは、クライアントバージョンが V1.10.14 以降の場合にのみ表示されます。
   単一マシンでの同時サブタスク数	ワーカー上の実行スレッドの数。デフォルト値: 5。実行を高速化するには、より大きな値を指定できます。ダウンストリームまたはデータベースが指定した値に耐えられない場合は、より小さな値を指定できます。
   サブタスクの失敗リトライ回数	タスクが失敗した場合のリトライ回数。デフォルト値: 0。
   サブタスクの失敗リトライ間隔	2 回の連続したリトライの間隔。単位: 秒。デフォルト値: 0。
   サブタスクのフェールオーバー戦略	ワーカーがタスクの実行に失敗して停止した後、新しいワーカーにタスクを分散するかどうかを指定します。スイッチをオンにすると、フェールオーバーがトリガーされたときにシステムがタスクを複数回実行する可能性があります。タスクのべき等性を実装する必要があります。 説明 このパラメーターは、クライアントバージョンが V1.8.13 以降の場合にのみ表示されます。
   マスターノードが実行に参加する	マスターノードがタスクの実行に参加するかどうかを指定します。タスクを実行するには、少なくとも 2 つのワーカーが利用可能である必要があります。非常に多数のタスクが存在する場合は、スイッチをオフにすることをお勧めします。 説明 このパラメーターは、クライアントバージョンが V1.8.13 以降の場合にのみ表示されます。

可視化機能

ジョブが実行された後、[実行リスト] ページでジョブを見つけ、[操作] 列の [詳細] をクリックしてタスクの実行詳細を表示できます。

• [サブタスクリスト] タブで、各タスクの実行ステータスを表示します。

  

• [サブタスクリスト] タブで、管理したいタスクを見つけ、[操作] 列の [ログ] をクリックして、各タスクのビジネスログ情報を表示し、実行結果を分析します。

  

• ジョブの実行レコードが利用可能になったら、[現在の実行詳細] タブの [ThreadDump] をクリックして、関連するマシンのスレッド実行ステータスを表示し、ジョブのエラー詳細を分析します。

  

• トレース分析が接続された後、[サブタスクリスト] タブのタスクの [TraceId] 列の値をクリックして、対応するトレースの詳細をクエリできます。詳細については、「トレース分析の統合」をご参照ください。