MapReduce model

SchedulerX MapReduce distributes batch data processing across connected workers through a simple Java API. Instead of importing data into a big data platform such as Hadoop or Spark, you process data in place -- splitting work into subtasks that run in parallel on your existing application nodes.

Compared to traditional big data frameworks, SchedulerX MapReduce offers:

No data migration: Process data directly in your databases without building an ETL pipeline to a separate cluster.
Fast startup: Tasks start in seconds, not minutes.
No infrastructure overhead: No additional storage or computing resources beyond your existing workers.
Simple programming model: Extend MapJobProcessor or MapReduceJobProcessor and implement two to three methods.

How it works

Map vs. MapReduce: pick the right processor

SchedulerX provides two processor interfaces. Choose based on whether you need to aggregate results after all subtasks complete.

Processor	Use case	Reduce phase
`MapJobProcessor`	Distribute work across workers. Each subtask runs independently.	No
`MapReduceJobProcessor`	Distribute work, then aggregate results or trigger downstream actions after all subtasks finish.	Yes

Note

If you do not need a reduce phase, use MapJobProcessor. The reduce phase caches all subtask results on the master node, which increases memory pressure.

Execution flow

All MapReduce jobs follow the same pattern:

SchedulerX invokes process() on the master node. The master detects that it is the root task.
The root task splits work into subtasks and calls map() to distribute them across workers.
Each worker receives a subtask, checks the task name, and runs the corresponding processing logic.
(MapReduce only) After all subtasks finish, SchedulerX calls reduce() on the master node to aggregate results.

You can call map() multiple times to create multi-level task trees, for example: database shards -> table shards -> page ranges.

API reference

MapJobProcessor methods

Method	Description	Required
`process(JobContext context)`	Entry point for task execution. Call `isRootTask(context)` to determine whether to distribute subtasks or process a subtask. Returns ProcessResult.	Yes
`map(List<? extends Object> taskList, String taskName)`	Distribute a batch of subtasks to workers. Can be called multiple times to create multi-level task trees. If `taskList` is empty, an error is returned. Returns ProcessResult.	Yes
`kill(JobContext context)`	Called when a job is stopped from the console. Implement cleanup or interruption logic here.	No

MapReduceJobProcessor methods

MapReduceJobProcessor includes all methods from MapJobProcessor, plus the following:

Method	Description	Required
`reduce(JobContext context)`	Runs on the master node after all subtasks complete. Use `context.getTaskStatuses()` and `context.getTaskResults()` to access subtask outcomes. Common uses: aggregate data, send notifications, or pass data to downstream workflow nodes.	Yes
`runReduceIfFail(JobContext context)`	Specifies whether to call the `reduce` method if a subtask fails. By default, `reduce` is executed even if a subtask fails.	No

How the reduce flow works:

Each subtask returns a result through return new ProcessResult(true, result), where result is a string (for example, an order ID or a computed value).
After all subtasks finish, SchedulerX calls reduce on the master node.
Inside reduce, call context.getTaskStatuses() for completion statuses and context.getTaskResults() for the result strings.

Important

The reduce phase caches all subtask results on the master node. To avoid out-of-memory errors, keep the number of subtasks and the size of each result value small. If you do not need result aggregation, use MapJobProcessor instead.

Limits

Constraint	Limit
Maximum task size	64 KB
Maximum `ProcessResult` result field size	1,000 bytes
Reduce memory impact	All subtask results are cached on the master node. Keep subtask count and result sizes small.
Idempotency requirement	SchedulerX may run a subtask more than once during a failover. Design subtask logic to be idempotent.

Create a MapReduce job

Log on to the SchedulerX console.
In the left-side navigation pane, click Task Management.
On the Task Management page, click Create task.
In the Create Task panel, set Task Type to Java and Execution Mode to MapReduce.
Configure the parameters in the Advanced Configuration section.

Advanced configuration parameters

Parameter	Description
distribution policy	How subtasks are assigned to workers. Polling Scheme (default): distributes an equal number of subtasks to each worker. Best when subtask processing times are similar. Optimal Load Policy: the master node monitors worker load and routes subtasks accordingly. Best when processing times vary across subtasks or workers. Requires agent version 1.10.3 or later.
Number of single-machine concurrent subtasks	Number of threads each worker uses to run subtasks. Default: 5. Increase for faster throughput; decrease if downstream systems cannot handle the load.
Number of failed retries of subtasks	Number of automatic retries for a failed subtask. Default: 0 (no retry).
Sub-task failure retry interval	Seconds between retry attempts. Default: 0.
Subtask Failover Strategy	When enabled, a subtask is reassigned to a different worker after the original worker fails to execute it and is stopped. Because this can cause duplicate execution, make sure subtask logic is idempotent. Requires agent version 1.8.12 or later.
The master node participates in the execution	When enabled, the master node also runs subtasks. Requires at least two workers. Disable this for jobs with a very large number of subtasks to reduce master node load. Requires agent version 1.8.12 or later.
Subtask distribution method	Push model: the master distributes subtasks evenly to workers. Pull model: workers pull subtasks on demand, which supports dynamic scale-up. The Wooden Bucket Theory is not applicable to this model. In pull mode, all pending subtasks are cached on the master node. To avoid memory issues, do not distribute more than 10,000 subtasks at a time.

For a complete list of parameters, see Advanced parameters for job management.

Examples

Scan a single table by ID range

Split a table into page-sized chunks using the minimum and maximum row IDs, then process each chunk on a separate worker.

How it works:

Query the ID range: SELECT MIN(id), MAX(id) FROM table_name.
Divide the range into pages. Each page becomes a subtask containing a startId and endId.
Each worker processes its assigned page: SELECT * FROM table_name WHERE id >= startId AND id < endId.

// Represents one page of work: a range of IDs to process.
class PageTask {
    private long startId;
    private long endId;

    public PageTask(long startId, long endId) {
        this.startId = startId;
        this.endId = endId;
    }

    public long getStartId() {
        return startId;
    }

    public long getEndId() {
        return endId;
    }
}


@Component
public class ScanSingleTableJobProcessor extends MapJobProcessor {

    @Autowired
    private XXXService xxxService;

    // Number of rows per subtask. Tune based on your row size and processing time.
    // Larger values mean fewer subtasks but higher retry cost on failure.
    // Smaller values mean more subtasks but increased scheduling overhead.
    private final int PAGE_SIZE = 500;

    @Override
    public ProcessResult process(JobContext context) throws Exception {
        // Use job parameters to pass the table name, so the same processor
        // class can handle multiple jobs with different tables.
        String tableName = context.getJobParameters();
        String taskName = context.getTaskName();
        Object task = context.getTask();

        if (isRootTask(context)) {
            // Root task: split the table into page-sized subtasks
            Pair<Long, Long> idPair = queryMinAndMaxId(tableName);
            long minId = idPair.getFirst();
            long maxId = idPair.getSecond();
            List<PageTask> tasks = Lists.newArrayList();
            int step = (int) ((maxId - minId) / PAGE_SIZE);
            if (step > 0) {
                for (long i = minId; i < maxId; i += step) {
                    tasks.add(new PageTask(i, (i + step > maxId ? maxId : i + step)));
                }
            } else {
                tasks.add(new PageTask(minId, maxId));
            }
            // map() distributes subtasks to workers. If the task list is empty,
            // an error is returned.
            return map(tasks, "PageTask");
        } else if (taskName.equals("PageTask")) {
            // Subtask: process one page of rows
            PageTask pageTask = (PageTask) task;
            long startId = pageTask.getStartId();
            long endId = pageTask.getEndId();
            List<Record> records = queryRecord(tableName, startId, endId);
            // Process records: update status, send notifications, etc.
            return new ProcessResult(true);
        }

        return new ProcessResult(false);
    }

    private Pair<Long, Long> queryMinAndMaxId(String tableName) {
        // TODO: SELECT MIN(id), MAX(id) FROM <table-name>
        return new Pair<Long, Long>(1L, 10000L);
    }

    private List<Record> queryRecord(String tableName, long startId, long endId) {
        List<Record> records = Lists.newArrayList();
        // TODO: SELECT * FROM <table-name> WHERE id >= <start-id> AND id < <end-id>
        return records;
    }

}

Scan a single table by hash bucket

When row IDs are non-sequential (for example, UUIDs or deleted rows create gaps), use a hash bucketing strategy instead of ID ranges.

Prerequisites: Add a bucket column to your table. When inserting a row, compute the bucket value (for example, order_id % 1024) and store it in this column. Index the bucket column.

How it works:

The root task creates one subtask per bucket (for example, 1,024 subtasks for 1,024 buckets).
Each worker queries all rows in its assigned bucket: SELECT * FROM table_name WHERE bucket = <bucket-id>.

@Component
public class ScanSingleTableJobProcessor extends MapJobProcessor {

    @Autowired
    private XXXService xxxService;

    @Override
    public ProcessResult process(JobContext context) throws Exception {
        String tableName = context.getJobParameters();
        String taskName = context.getTaskName();
        Object task = context.getTask();

        if (isRootTask(context)) {
            // Root task: create one subtask per bucket.
            // Adjust the bucket count to match your table's bucketing scheme.
            List<Integer> tasks = Lists.newArrayList();
            for (int i = 0; i < 1024; i++) {
                tasks.add(i);
            }
            return map(tasks, "BucketTask");
        } else if (taskName.equals("BucketTask")) {
            // Subtask: process all rows in one bucket
            int bucketId = (int) task;
            List<Record> records = queryRecord(tableName, bucketId);
            // Process records
            return new ProcessResult(true);
        }

        return new ProcessResult(false);
    }

    private List<Record> queryRecord(String tableName, int bucketId) {
        List<Record> records = Lists.newArrayList();
        // TODO: SELECT * FROM <table-name> WHERE bucket = <bucket-id>
        return records;
    }

}

Scan sharded databases and tables

For data spread across multiple database shards and table shards, use multi-level mapping to break the work down progressively: database shard -> table shard -> page range within each table.

How it works:

The root task lists all database shards and distributes them as level-1 subtasks.
Each level-1 subtask lists the table shards in its database and distributes them as level-2 subtasks.
Each level-2 subtask splits a table shard into page-sized ranges and distributes them as leaf subtasks.
Each leaf subtask queries and processes one page of rows.

Note

Multi-level mapping is a key differentiator of SchedulerX MapReduce. Each call to map() creates a new level in the task tree. Use descriptive task names (such as DbTask, TableTask, PageTask) to route subtasks to the correct processing branch.

// Represents one page of work within a specific table shard
class PageTask {
    private String tableName;
    private long startId;
    private long endId;

    public PageTask(String tableName, long startId, long endId) {
        this.tableName = tableName;
        this.startId = startId;
        this.endId = endId;
    }

    public String getTableName() {
        return tableName;
    }

    public long getStartId() {
        return startId;
    }

    public long getEndId() {
        return endId;
    }
}


@Component
public class ScanShardingTableJobProcessor extends MapJobProcessor {

    @Autowired
    private XXXService xxxService;

    private final int PAGE_SIZE = 500;

    @Override
    public ProcessResult process(JobContext context) throws Exception {
        String taskName = context.getTaskName();
        Object task = context.getTask();

        if (isRootTask(context)) {
            // Level 1: distribute database shards
            List<String> dbList = getDbList();
            return map(dbList, "DbTask");
        } else if (taskName.equals("DbTask")) {
            // Level 2: distribute table shards within a database
            String dbName = (String) task;
            List<String> tableList = getTableList(dbName);
            return map(tableList, "TableTask");
        } else if (taskName.equals("TableTask")) {
            // Level 3: split a table shard into page-sized subtasks
            String tableName = (String) task;
            Pair<Long, Long> idPair = queryMinAndMaxId(tableName);
            long minId = idPair.getFirst();
            long maxId = idPair.getSecond();
            List<PageTask> tasks = Lists.newArrayList();
            int step = (int) ((maxId - minId) / PAGE_SIZE);
            if (step > 0) {
                for (long i = minId; i < maxId; i += step) {
                    tasks.add(new PageTask(tableName, i, (i + step > maxId ? maxId : i + step)));
                }
            } else {
                tasks.add(new PageTask(tableName, minId, maxId));
            }
            return map(tasks, "PageTask");
        } else if (taskName.equals("PageTask")) {
            // Leaf task: process one page of rows
            PageTask pageTask = (PageTask) task;
            String tableName = pageTask.getTableName();
            long startId = pageTask.getStartId();
            long endId = pageTask.getEndId();
            List<Record> records = queryRecord(tableName, startId, endId);
            // Process records
            return new ProcessResult(true);
        }

        return new ProcessResult(false);
    }

    private List<String> getDbList() {
        List<String> dbList = Lists.newArrayList();
        // TODO: Return the list of database shards
        return dbList;
    }

    private List<String> getTableList(String dbName) {
        List<String> tableList = Lists.newArrayList();
        // TODO: Return the list of table shards for the specified database
        return tableList;
    }

    private Pair<Long, Long> queryMinAndMaxId(String tableName) {
        // TODO: SELECT MIN(id), MAX(id) FROM <table-name>
        return new Pair<Long, Long>(1L, 10000L);
    }

    private List<Record> queryRecord(String tableName, long startId, long endId) {
        List<Record> records = Lists.newArrayList();
        // TODO: SELECT * FROM <table-name> WHERE id >= <start-id> AND id < <end-id>
        return records;
    }

}

Distribute messages with reduce

Process a batch of messages across workers, then run a reduce step to report the outcome.

@Component
public class TestMapReduceJobProcessor extends MapReduceJobProcessor {

    @Override
    public ProcessResult process(JobContext context) throws Exception {
        String taskName = context.getTaskName();
        int dispatchNum = 50;

        if (isRootTask(context)) {
            // Root task: create 50 message subtasks
            List<String> msgList = Lists.newArrayList();
            for (int i = 0; i <= dispatchNum; i++) {
                msgList.add("msg_" + i);
            }
            return map(msgList, "Level1Dispatch");
        } else if (taskName.equals("Level1Dispatch")) {
            // Subtask: process one message
            String task = (String) context.getTask();
            System.out.println(task);
            return new ProcessResult(true);
        }

        return new ProcessResult(false);
    }

    @Override
    public ProcessResult reduce(JobContext context) throws Exception {
        // Called after all subtasks complete.
        // Warning: all subtask results are cached in memory on the master node.
        // For large-scale jobs (10,000+ subtasks), consider whether you truly
        // need reduce, or if MapJobProcessor is sufficient.
        return new ProcessResult(true, "TestMapReduceJobProcessor.reduce");
    }

}

Distribute messages and aggregate results in reduce

Extend the previous example to collect each subtask's result and aggregate them in the reduce phase.

@Component
public class TestMapReduceJobProcessor extends MapReduceJobProcessor {

    @Override
    public ProcessResult process(JobContext context) throws Exception {
        String taskName = context.getTaskName();
        int dispatchNum = 50;

        // Optionally override the number of messages through job parameters
        if (context.getJobParameters() != null) {
            dispatchNum = Integer.valueOf(context.getJobParameters());
        }

        if (isRootTask(context)) {
            List<String> msgList = Lists.newArrayList();
            for (int i = 0; i <= dispatchNum; i++) {
                msgList.add("msg_" + i);
            }
            return map(msgList, "Level1Dispatch");
        } else if (taskName.equals("Level1Dispatch")) {
            String task = (String) context.getTask();
            Thread.sleep(2000);
            // Return the message as the subtask result so reduce can collect it.
            // The result string must not exceed 1,000 bytes.
            return new ProcessResult(true, task);
        }

        return new ProcessResult(false);
    }

    @Override
    public ProcessResult reduce(JobContext context) throws Exception {
        // context.getTaskResults() returns a Map<Long, String>:
        //   key   = subtask ID
        //   value = the result string from ProcessResult(true, result)
        for (Entry<Long, String> result : context.getTaskResults().entrySet()) {
            System.out.println("taskId:" + result.getKey() + ", result:" + result.getValue());
        }
        return new ProcessResult(true, "TestMapReduceJobProcessor.reduce");
    }
}

Best practices

Choose the simplest processor: Use MapJobProcessor unless you need result aggregation. The reduce phase adds memory overhead on the master node.
Design for idempotency: Subtasks may run more than once during failover. Use unique IDs or idempotency keys to prevent duplicate side effects.
Tune PAGE_SIZE for your workload: Larger pages mean fewer subtasks but higher retry cost on failure. Smaller pages improve parallelism but increase scheduling overhead.
Handle errors from map(): If the task list is empty, map() returns an error. Validate your task list before calling map() in production code.
Monitor master node memory: When using reduce or pull mode, subtask data is cached on the master node. In pull mode, do not distribute more than 10,000 subtasks at a time.
Use descriptive task names: When building multi-level task trees, use names like DbTask, TableTask, and PageTask to route subtasks to the correct processing branch.

For architecture details, scheduling internals, and performance tuning guidance, see Principles and best practices for SchedulerX 2.0 distributed computing.