ToB企服应用市场:ToB评测及商务社交产业平台

标题: PoweJob高级特性-MapReduce完整示例 [打印本页]

作者: 忿忿的泥巴坨 时间: 2022-8-24 16:04
标题: 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过程获取每个子任务的执行结果汇总累加，返回结果值。
源代码

[code]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 */@Componentpublic 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 jobParams = Splitter.on("&").withKeyValueSeparator("=").split(context.getJobParams()); Integer batchSize = Integer.parseInt(jobParams.getOrDefault("batchSize", "100")); Integer batchNum = Integer.parseInt(jobParams.getOrDefault("batchNum", "10")); List 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

欢迎光临 ToB企服应用市场:ToB评测及商务社交产业平台 (https://dis.qidao123.com/)