Microservices Engine:Visual MapReduce model

Step 1: Define the task object with labels

Visual MapReduce inherits all interfaces from the standard MapReduce model. To add custom business labels (such as account name, product code, or city) to the console subtask list, implement the com.alibaba.schedulerx.worker.processor.BizSubTask interface on your task objects.

Create a class that implements BizSubTask so that each subtask displays its 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 for this subtask.
     * Labels appear as filterable columns in the console subtask list.
     */
    @Override
    public Map<String, String> labelMap() {
        Map<String, String> labelMap = new HashMap();
        labelMap.put("Account name", name);
        return labelMap;
    }
}

After you implement BizSubTask, the subtask list displays label columns. You can filter subtasks by label to locate a specific account.

Step 2: Implement the job processor

Extend MapReduceJobProcessor to define the root task logic (build the subtask list) and the subtask logic (process each account).

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 the list of subtask objects.
             *  In production, load these objects from your data source. Examples:
             *  1. Query unprocessed customer accounts from the database.
             *  2. Build a region table to distribute jobs by province, city, or district.
             *  3. Group tasks by business category (e.g., appliances, groceries, food).
             *  4. Partition tasks by time period (e.g., January, February).
             */
            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 task object and process it.
             */
            ParallelAccountInfo obj = (ParallelAccountInfo) context.getTask();
            // Process the account
            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 trigger it. See the next section for console configuration steps.

Subtask list

View the execution status of each subtask on the Subtask list tab. The list shows status, custom labels, and operation options for every subtask in the job. Use this view to get a quick overview of overall job progress and identify failed or stuck subtasks.

Subtask logs

On the Subtask list tab, find the target subtask and click Log in the Operation column to view its business log. Use subtask logs to inspect the processing output or trace errors for a specific subtask without searching through centralized log files.

Thread dump

On the Current execution details tab, click ThreadDump to view the thread execution status of the relevant machine. Use thread dumps to diagnose subtasks that are stuck, blocked on I/O, or running longer than expected.

Trace analysis

If tracing analysis is configured, click the TraceId value for a subtask on the Subtask list tab to view its distributed trace. Use trace analysis to identify latency bottlenecks across service calls within a subtask. For setup instructions, see Integrate tracing analysis.

Size and quantity

Memory

When the reduce method is used, all subtask results are cached on the master node, which increases memory pressure. Keep subtask counts and result sizes small. If you do not need reduce, extend MapJobProcessor instead of MapReduceJobProcessor.

Idempotence

If a failover occurs, SchedulerX may run a subtask more than once. Make sure your subtask logic is idempotent.

SDK version

• SDK versions 1.12.2 and earlier contain security vulnerabilities. Upgrade to a version later than 1.12.2.

• Visual MapReduce requires SchedulerX Professional Edition.