PoweJob高级特性-MapReduce完整示例

由于网上搜索 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;         }     }  }   

执行日志如图
PoweJob高级特性-MapReduce完整示例
执行结果如图
PoweJob高级特性-MapReduce完整示例

发表评论

相关文章