由于网上搜索 PowerJob MapReduce 都是设计原理，demo也展示个空壳子，没有演示Map到Reduce结果怎么传递，对于没有MR开发经验的人来说并没有什么帮助，所以这里写了一个有完整计算意义的demo供参考。
代码功能：
实现一个sum累加。
任务输入参数：
batchSize=100&batchNum=10，
其中batchSize表示每个子任务大小，这里就是一个子任务负责100个数据累加。
batchNum表示批次大小，也就是本次分发为10个子任务来完成。
执行过程就是：Map过程是将本次任务划分为10个子任务，每个子任务分别完成1累加到100，101累加到201，...，以此类推。Reduce过程获取每个子任务的执行结果汇总累加，返回结果值。
源代码
package org.example.demo; import com.google.common.base.Splitter; import com.google.common.collect.Lists; import org.apache.commons.lang3.math.NumberUtils; import org.springframework.stereotype.Component; import tech.powerjob.common.serialize.JsonUtils; import tech.powerjob.worker.core.processor.ProcessResult; import tech.powerjob.worker.core.processor.TaskContext; import tech.powerjob.worker.core.processor.TaskResult; import tech.powerjob.worker.core.processor.sdk.MapReduceProcessor; import tech.powerjob.worker.log.OmsLogger; import java.io.Serializable; import java.util.List; import java.util.Map; /** * 控制台参数 batchSize=100&batchNum=10 * @author zhengqian * @date 2022.05.30 */ @Component public class MRSumProcessor implements MapReduceProcessor { @Override public ProcessResult process(TaskContext context) throws Exception { OmsLogger omsLogger = context.getOmsLogger(); System.out.println("============== TestMapReduceProcessor#process =============="); System.out.println("isRootTask:" + isRootTask()); System.out.println("taskContext:" + JsonUtils.toJSONString(context)); if (isRootTask()) { System.out.println("==== MAP ===="); omsLogger.info("[DemoMRProcessor] start root task~"); // 根据控制台参数获取MR批次及子任务大小 Map<String, String> jobParams = Splitter.on("&").withKeyValueSeparator("=").split(context.getJobParams()); Integer batchSize = Integer.parseInt(jobParams.getOrDefault("batchSize", "100")); Integer batchNum = Integer.parseInt(jobParams.getOrDefault("batchNum", "10")); List<SubTaskParam> subTasks = Lists.newLinkedList(); for (int j = 0; j < batchNum; j++) { subTasks.add(new SubTaskParam(j * batchSize + 1, (j + 1) * batchSize)); map(subTasks, "INFO"); subTasks.clear(); } omsLogger.info("[DemoMRProcessor] map success~"); return new ProcessResult(true, "MAP_SUCCESS"); } else if (context.getTaskName().equals("INFO")) { // 子任务执行 SubTaskParam subTaskParam = (SubTaskParam) context.getSubTask(); omsLogger.info(subTaskParam.toString()); long sum = 0L; for (int x = subTaskParam.getStart(); x <= subTaskParam.getEnd(); x++) { sum += x; } omsLogger.info("[DemoMRProcessor] start={}, end={}, sum={}", subTaskParam.getStart(), subTaskParam.getEnd(), sum); return new ProcessResult(true, String.valueOf(sum)); } return new ProcessResult(false); } @Override public ProcessResult reduce(TaskContext context, List<TaskResult> taskResults) { log.info("================ MapReduceProcessorDemo#reduce ================"); log.info("TaskContext: {}", JsonUtils.toJSONString(context)); log.info("List<TaskResult>: {}", JsonUtils.toJSONString(taskResults)); context.getOmsLogger().info("MapReduce job finished, result is {}.", taskResults); long sum = 0L; for (TaskResult taskResult : taskResults) { String result = taskResult.getResult(); if (NumberUtils.isDigits(result)) { sum += Long.parseLong(result); } } return new ProcessResult(true, sum + ": " + sum); } private static class SubTaskParam implements Serializable { private int start; private int end; public SubTaskParam() {} public SubTaskParam(int start, int end) { this.start = start; this.end = end; } public int getStart() { return start; } public void setStart(int start) { this.start = start; } public int getEnd() { return end; } public void setEnd(int end) { this.end = end; } @Override public String toString() { return start + ":" + end; } } } 执行日志如图
执行结果如图
