可视化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平均分配等量子任务，适用于每个子任务处理耗时基本一致的场景。 WorkerLoad最优策略：由主节点自动感知Worker节点的负载情况，适用于子任务和Worker机器处理耗时有较大差异的场景。 说明 客户端版本为1.10.14及以上。
   子任务单机并发数	即单机执行线程数，默认为5。如需加快执行速度，可以调大该值。如果下游或者数据库无法承接，可适当调小。
   子任务失败重试次数	子任务失败会自动重试，默认为0。
   子任务失败重试间隔	子任务失败重试间隔，单位：秒，默认为0。
   子任务failover策略	当执行节点宕机下线后，是否将子任务重新分发给其他机器执行。开启该配置后，发生failover时，子任务可能会重复执行，需自行做好幂等。 说明 客户端版本为1.8.13及以上。
   主节点参与执行	主节点是否参与子任务执行。在线可运行Worker数量必须不低于2台，在子任务数量特别大时，推荐关闭该参数。 说明 客户端版本为1.8.13及以上。

可视化能力

任务执行后，您可以在执行列表页面，单击详情查看对应子任务的详细执行信息。

• 在子任务列表页签查看每个子任务处理的状态。

  

• 在子任务列表页签，单击子任务操作列的日志，可以查看每个子任务运行的业务日志信息，分析执行状态结果。

  

• 任务执行记录在运行中时，在当前执行详情页签，单击查看堆栈，可以查看对应机器处理线程运行中的情况，分析当前任务运行异常情况。

  

• 在子任务列表页签，当接入链路追踪后，单击对应的TraceId，可以查询每个子任务的执行调用链路。具体操作，请参见如何接入链路追踪。