[feat_1193][taier-jobGraph] Expand the new scheduleJobs T+1 allocation strategy algorithm #1193

taier-common/src/main/java/com/dtstack/taier/common/enums/EScheduleJobDistributeType.java

@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.dtstack.taier.common.enums;
+
+/**
+ * 周期实例分配策略
+ * @author xingyi
+ */
+public enum EScheduleJobDistributeType {
+
+    /**
+     * 历史默认策略，通过构建schedule_engine_job_cache 进行初始化分配，依据数据总量+节点负载 来分配
+     */
+    DEFAULT(0),
+
+    /**
+     * 依据ScheduleJob -> taskType_cycTime 策略进行平均分配，不关注schedule_engine_job_cache
+     */
+    TASK_TYPE_CYCTIME(1);
+
+    private int value;
+
+    EScheduleJobDistributeType(int value) {
+        this.value = value;
+    }
+
+    public int getValue() {
+        return value;
+    }
+
+    public static EScheduleJobDistributeType getDistributeType(int value) {
+        for (EScheduleJobDistributeType type : EScheduleJobDistributeType.values()) {
+            if (type.value == value) {
+                return type;
+            }
+        }
+        return EScheduleJobDistributeType.DEFAULT;
+    }
+}

taier-common/src/main/java/com/dtstack/taier/common/env/EnvironmentContext.java

@@ -687,6 +687,15 @@ public long getPrometheusPushGatewayInterval() {
         return Long.parseLong(environment.getProperty("taier.monitor.metrics.prometheus.pushgateway.interval", "60"));
     }
 
+    /**
+     * 周期实例生成策略
+     * 0: 默认策略
+     * 1: 按照taskType_cycTime 进行均分配策略
+     */
+    public int getJobGraphDistributeType() {
+        return Integer.parseInt(environment.getProperty("engine.job.graph.distribute.type", "0"));
+    }
+
 
 
 

taier-scheduler/src/main/java/com/dtstack/taier/scheduler/server/builder/CycleJobBuilder.java

@@ -19,13 +19,15 @@
 package com.dtstack.taier.scheduler.server.builder;
 
 import com.dtstack.taier.common.enums.Deleted;
+import com.dtstack.taier.common.enums.EScheduleJobDistributeType;
 import com.dtstack.taier.common.enums.EScheduleStatus;
 import com.dtstack.taier.common.enums.EScheduleType;
 import com.dtstack.taier.common.exception.TaierDefineException;
 import com.dtstack.taier.dao.domain.ScheduleTaskShade;
 import com.dtstack.taier.pluginapi.util.DateUtil;
 import com.dtstack.taier.pluginapi.util.RetryUtil;
 import com.dtstack.taier.scheduler.server.ScheduleJobDetails;
+import com.dtstack.taier.scheduler.server.distribute.ScheduleJobDistributeContext;
 import com.dtstack.taier.scheduler.service.JobGraphTriggerService;
 import com.dtstack.taier.scheduler.utils.JobExecuteOrderUtil;
 import com.google.common.collect.Lists;
@@ -98,6 +100,8 @@ public void buildTaskJobGraph(String triggerDay) {
             CountDownLatch ctl = new CountDownLatch(totalBatch);
             AtomicJobSortWorker sortWorker = new AtomicJobSortWorker();
 
+            ScheduleJobDistributeContext distributeContext = new ScheduleJobDistributeContext();
+
             // 3. 查询db多线程生成周期实例
             Long startId = 0L;
             for (int i = 0; i < totalBatch; i++) {
@@ -124,7 +128,7 @@ public void buildTaskJobGraph(String triggerDay) {
                                     List<ScheduleJobDetails> scheduleJobDetails = RetryUtil.executeWithRetry(() -> buildJob(batchTaskShade, triggerDay, sortWorker),
                                             environmentContext.getBuildJobErrorRetry(), 200, false);
                                     // 插入周期实例
-                                    savaJobList(scheduleJobDetails);
+                                    savaJobList(scheduleJobDetails, distributeContext);
                                 } catch (Throwable e) {
                                     LOGGER.error("build task failure taskId:{}", batchTaskShade.getTaskId(), e);
                                 }
@@ -166,7 +170,7 @@ private void clearInterruptJob(Timestamp triggerDay) {
      * @param scheduleJobDetails 实例详情
      */
     @Transactional(rollbackFor = Exception.class)
-    public void savaJobList(List<ScheduleJobDetails> scheduleJobDetails) {
+    public void savaJobList(List<ScheduleJobDetails> scheduleJobDetails, ScheduleJobDistributeContext distributeContext) {
         List<ScheduleJobDetails> savaJobDetails = Lists.newArrayList();
         for (ScheduleJobDetails scheduleJobDetail : scheduleJobDetails) {
             savaJobDetails.add(scheduleJobDetail);
@@ -177,7 +181,8 @@ public void savaJobList(List<ScheduleJobDetails> scheduleJobDetails) {
             }
         }
 
-        scheduleJobService.insertJobList(savaJobDetails, getType());
+        EScheduleJobDistributeType distributeType = EScheduleJobDistributeType.getDistributeType(environmentContext.getJobGraphDistributeType());
+        scheduleJobService.insertJobList(savaJobDetails, EScheduleType.NORMAL_SCHEDULE.getType(), distributeType, distributeContext);
     }
 
     /**

taier-scheduler/src/main/java/com/dtstack/taier/scheduler/server/distribute/DefaultScheduleJobDistributeStrategy.java

@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.dtstack.taier.scheduler.server.distribute;
+
+import com.dtstack.taier.common.enums.EScheduleJobDistributeType;
+import com.dtstack.taier.common.exception.TaierDefineException;
+import com.dtstack.taier.scheduler.server.JobPartitioner;
+import com.dtstack.taier.scheduler.server.ScheduleJobDetails;
+import com.google.common.collect.Maps;
+import org.apache.commons.collections.MapUtils;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * 历史默认策略，按调度节点当前负载计算每个节点应接收的实例数。
+ * @author xingyi
+ */
+@Component
+public class DefaultScheduleJobDistributeStrategy implements ScheduleJobDistributeStrategy {
+
+    @Autowired
+    private JobPartitioner jobPartitioner;
+
+    @Override
+    public EScheduleJobDistributeType distributeType() {
+        return EScheduleJobDistributeType.DEFAULT;
+    }
+
+    @Override
+    public Map<ScheduleJobDetails, String> distribute(List<ScheduleJobDetails> scheduleJobDetails,
+                                                      Integer scheduleType,
+                                                      ScheduleJobDistributeContext distributeContext) {
+        Map<ScheduleJobDetails, String> jobNodeMap = Maps.newHashMap();
+        Map<String, Integer> nodeJobSize = jobPartitioner.computeBatchJobSize(scheduleType, scheduleJobDetails.size());
+        if (MapUtils.isEmpty(nodeJobSize)) {
+            throw new TaierDefineException("No available node to distribute schedule jobs");
+        }
+
+        Iterator<ScheduleJobDetails> batchJobIterator = scheduleJobDetails.iterator();
+        for (Map.Entry<String, Integer> nodeJobSizeEntry : nodeJobSize.entrySet()) {
+            String nodeAddress = nodeJobSizeEntry.getKey();
+            int nodeSize = nodeJobSizeEntry.getValue();
+            while (nodeSize > 0 && batchJobIterator.hasNext()) {
+                jobNodeMap.put(batchJobIterator.next(), nodeAddress);
+                nodeSize--;
+            }
+        }
+        return jobNodeMap;
+    }
+}

taier-scheduler/src/main/java/com/dtstack/taier/scheduler/server/distribute/ScheduleJobDistributeContext.java

@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.dtstack.taier.scheduler.server.distribute;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * 周期实例分配上下文，用于在分批生成实例时共享策略状态。
+ * @author xingyi
+ */
+public class ScheduleJobDistributeContext {
+
+    private final Map<String, Long> globalNodeLoadCache = new ConcurrentHashMap<>();
+
+    private final Map<String, Map<String, Long>> globalGroupNodeLoadCache = new ConcurrentHashMap<>();
+
+    public Map<String, Long> getGlobalNodeLoadCache() {
+        return globalNodeLoadCache;
+    }
+
+    public Map<String, Map<String, Long>> getGlobalGroupNodeLoadCache() {
+        return globalGroupNodeLoadCache;
+    }
+}

taier-scheduler/src/main/java/com/dtstack/taier/scheduler/server/distribute/ScheduleJobDistributeStrategy.java

@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.dtstack.taier.scheduler.server.distribute;
+
+import com.dtstack.taier.common.enums.EScheduleJobDistributeType;
+import com.dtstack.taier.scheduler.server.ScheduleJobDetails;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * 周期实例节点分配策略。
+ * @author xingyi
+ */
+public interface ScheduleJobDistributeStrategy {
+
+    EScheduleJobDistributeType distributeType();
+
+    Map<ScheduleJobDetails, String> distribute(List<ScheduleJobDetails> scheduleJobDetails,
+                                               Integer scheduleType,
+                                               ScheduleJobDistributeContext distributeContext);
+}