Visual MapReduce model

Step 1: Define subtask objects

Each subtask object represents one unit of parallel work. To display custom business labels in the console task list, implement the com.alibaba.schedulerx.worker.processor.BizSubTask interface and its labelMap() method.

The following example defines a bank account object with a custom label that displays the account name in the console:

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;
    }

    /**
     * Return label key-value pairs that appear in the console task list.
     */
    @Override
    public Map<String, String> labelMap() {
        Map<String, String> labelMap = new HashMap();
        labelMap.put("Account name", name);
        return labelMap;
    }
}

After you implement this interface, each subtask shows its label in the Subtask list tab. Filter or search subtasks by label to quickly locate a specific account.

Step 2: Implement the processor

Extend com.alibaba.schedulerx.worker.processor.MapReduceJobProcessor and implement the process method. In the root task, build the list of subtask objects and call map() to distribute them. In each subtask, retrieve the object and run the business logic.

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();
            /**
             * Root task: build parallel subtask objects.
             * In practice, load objects based on your requirements. Examples:
             *  1. Load unprocessed customer accounts from a database.
             *  2. Build a region table to distribute work by province, city, or district.
             *  3. Classify tasks by business category (appliance, daily necessity, food).
             *  4. Partition by time period (January, February, and so on).
             */
            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 {
            // Subtask: retrieve the object and run business logic.
            ParallelAccountInfo obj = (ParallelAccountInfo) context.getTask();
            logger.info("Process task information: {}", JSON.toJSONString(obj));
            return new ProcessResult(true);
        }
    }
}

After you develop and deploy the job, create a scheduled task in the SchedulerX console to run it. See Configure a visual MapReduce job in the console for details.

View subtask status

On the Subtask list tab, view the execution status of each subtask, including custom labels.

View subtask logs

On the Subtask list tab, find the target subtask and click Log in the Operation column to view its business log output.

Analyze thread dumps

On the Current execution details tab, click ThreadDump to inspect thread execution on the relevant machine. Use this to diagnose stuck or slow subtasks.

Trace subtask execution

If Tracing Analysis is enabled, click the TraceId value on the Subtask list tab to view the distributed trace for that subtask. For setup instructions, see Integrate tracing analysis.

Hard limits

Developer responsibilities

• Idempotence. If a failover is triggered, SchedulerX may run a subtask more than once. Make sure that each subtask is idempotent.

• Memory on the master node. If the job uses the reduce method, the results of all subtasks are cached on the master node. Keep the subtask count and result size small to avoid memory pressure. If reduce is not needed, extend MapJobProcessor instead of MapReduceJobProcessor.