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;
        }
    }

}

執行日誌如圖

執行結果如圖