Visual MapReduce model

Batch processing of accounts

This example shows how to batch-process a list of bank accounts. Each account is handled as an independent subtask that runs in parallel across the cluster. Each subtask in the execution list displays its corresponding account information. This lets you easily track the processing status and execution details of each account. The following sections provide sample code.

1. Define a custom subtask object for account information. To display a label for each subtask object, implement the com.alibaba.schedulerx.worker.processor.BizSubTask interface and its labelMap method.

   Code

   public class ParallelAccountInfo implements BizSubTask {
   
       /**
        * Primary key
        */
       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;
       }
   
       /**
        * Implement the labelMap method to set the label for the subtask.
        * @return
        */
       @Override
       public Map<String, String> labelMap() {
           Map<String, String> labelMap = new HashMap();
           labelMap.put("Account Name", name);
           return labelMap;
       }
   }

   After the subtask object implements the interface, the subtask list can display a unique label for each object, such as the account name in this example. This lets you distinguish the processing status of each account. You can also search for subtasks by label.

   

2. Create a processor for the account business logic. This processor handles the logic for a single account and inherits from com.alibaba.schedulerx.worker.processor.MapReduceJobProcessor.

   Code

   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("Building a list of subtasks for parallel processing...");
               List<ParallelAccountInfo> list = new LinkedList();
               /**
                * If it is the root task, build a list of subtask objects for parallel processing.
                * In a real-world scenario, you can load subtask objects based on your business needs. These objects must implement the BizSubTask interface.
                * Example use cases:
                *  1. Load unprocessed customer accounts from a database.
                *  2. Create a list of provinces, cities, and regions to distribute tasks by area.
                *  3. Classify subtasks based on business labels, such as 'electronics', 'daily goods', or 'food'.
                *  4. Classify subtasks by time, such as by month (January, February, etc.).
                */
               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 {
               /**
                * For non-root tasks, you can retrieve the corresponding subtask object and perform business processing.
                */
               ParallelAccountInfo obj = (ParallelAccountInfo)context.getTask();
               // Process the business logic for the retrieved subtask object.
               // do something
               logger.info("Processing subtask: {}", JSON.toJSONString(obj));
               return new ProcessResult(true);
           }
       }
   }

   After you develop and deploy the task, configure and run it as a scheduled task in the console. For more information, see Procedure.

Configure the task

1. Log on to the SchedulerX console. In the left-side navigation pane, click task management.

2. On the task management page, click Create Task.

3. In the Create Task panel, select Visual MapReduce from the Execution Mode drop-down list.

   

4. In the advanced configuration section, set the following parameters. For information about other parameters, see Advanced parameters for task management.

   Parameter	Description
   Distribution policy	Polling Scheme (Default): Distributes an equal number of subtasks to each worker. This strategy is suitable for scenarios where subtasks have similar processing times. Worker load optimal strategy: The master node automatically detects the load on each worker and distributes subtasks accordingly. This strategy is suitable for scenarios where subtasks and workers have significantly different processing times. Note Requires client version 1.10.14 or later.
   Subtask concurrency per worker	The number of execution threads per worker. The default is 5. You can increase this value to speed up execution. If downstream systems or databases cannot handle the load, you can decrease this value.
   Subtask retry attempts	The number of times a subtask is automatically retried after a failure. The default is 0.
   Subtask retry interval	The interval between retry attempts for a failed subtask, in seconds. The default is 0.
   Subtask failover strategy	Specifies whether to redistribute subtasks to other machines if an execution node crashes or goes offline. If you enable this option, subtasks may be executed more than once during a failover. You must ensure that your business logic is idempotent. Note Requires client version 1.8.13 or later.
   Master participation	Specifies whether the master node participates in subtask execution. This requires at least two online workers. Disable this option when the number of subtasks is very large. Note Requires client version 1.8.13 or later.

Monitor and debug subtasks

After a task is executed, go to the Execution List page and click Details to view detailed information about its subtasks.

• On the subtask list tab, you can view the processing status of each subtask.

  

• On the subtask list tab, find the target subtask and click Log in the Operation column to view its business logs and analyze the execution results.

  

• While a task is running, go to the Current execution details tab and click thread dump to inspect the running threads on the corresponding machine. This helps analyze exceptions in the current task.

  

• On the subtask list tab, if you have integrated Tracing Analysis, you can click a TraceId to query the distributed trace for that subtask. For more information, see Integrate Tracing Analysis.