SchedulerX：可視化MapReduce模型

賬戶批量處理

案例描述：對一批銀行賬戶進行批量處理，每個帳號作為獨立的子任務在整個叢集中進行全域平行處理，並且每一個子任務在執行列表中需要顯示其對應的賬戶資訊以便查看，可以方便的掌握每一個帳號地處理狀態及其執行詳細資料。如下將提供相應demo代碼供參考使用。

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("戶名", 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("構建並行計算的子任務列表...");
               List<ParallelAccountInfo> list = new LinkedList();
               /**
                *  判斷如果是rootTask的情況下，構建並行計運算元任務對象列表
                *  在實際業務情境中，使用者可自行根據業務情境載入子任務對象且該業務對象實現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 {
               /**
                * 非rootTask，使用者可以擷取對應的子任務資訊進行相應的業務處理
                */
               ParallelAccountInfo obj = (ParallelAccountInfo)context.getTask();
               // 針對擷取的 obj子任務資訊，進行商務邏輯處理
               // do something
               logger.info("處理子任務資訊：{}", JSON.toJSONString(obj));
               return new ProcessResult(true);
           }
       }
   }

   完成任務開發部署後，在控制台配置相應定時任務運行，請參見操作步驟。

任務配置

1. 登入分布式任務調度平台，在左側導覽列，單擊任務管理。

2. 在任務管理頁面，單擊建立任務。

3. 在建立任務面板，執行模式下拉式清單選擇可視化MapReduce。

   

4. 在進階配置地區配置相關資訊。其他配置項，請參見任務管理進階配置參數說明。

   配置項	說明
   分發策略	輪詢策略（預設）：每個Worker平均分配等量子任務，適用於每個子任務處理耗時基本一致的情境。 負載最優策略：由主節點自動感知Worker節點的負載情況，適用於子任務和Worker機器處理耗時有較大差異的情境。 說明 用戶端版本為1.10.14及以上。
   子任務單機並發數	即單機執行線程數，預設為5。如需加快執行速度，可以調大該值。如果下遊或者資料庫無法承接，可適當調小。
   子任務失敗重試次數	子任務失敗會自動重試，預設為0。
   子任務失敗稍候再試	子任務失敗稍候再試，單位：秒，預設為0。
   子任務failover策略	當執行節點宕機下線後，是否將子任務重新分發給其他機器執行。開啟該配置後，發生failover時，子任務可能會重複執行，需自行做好等冪。 說明 用戶端版本為1.8.13及以上。
   主節點參與執行	主節點是否參與子任務執行。線上可運行Worker數量必須不低於2台，在子任務數量特別大時，推薦關閉該參數。 說明 用戶端版本為1.8.13及以上。

可視化能力

任務執行後，您可以在執行列表頁面，單擊詳情查看對應子任務的詳細執行資訊。

• 在子任務列表頁簽查看每個子任務處理的狀態。

  

• 在子任務列表頁簽，單擊子任務操作列的日誌，可以查看每個子任務啟動並執行業務日誌資訊，分析執行狀態結果。

  

• 任務執行記錄在運行中時，在當前執行詳情頁簽，單擊查看堆棧，可以查看對應機器處理線程運行中的情況，分析當前任務運行異常情況。

  

• 在子任務列表頁簽，當接入鏈路追蹤後，單擊對應的TraceId，可以查詢每個子任務的執行調用鏈路。具體操作，請參見如何接入鏈路追蹤。