[
https://issues.apache.org/jira/browse/KYLIN-6087?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18098381#comment-18098381
]
ASF GitHub Bot commented on KYLIN-6087:
---------------------------------------
fishcus commented on code in PR #2347:
URL: https://github.com/apache/kylin/pull/2347#discussion_r3635865925
##########
src/data-loading-service/src/main/java/org/apache/kylin/rest/scheduler/AutoBuildSegmentScheduler.java:
##########
@@ -0,0 +1,248 @@
+/*
+ * 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 org.apache.kylin.rest.scheduler;
+
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
+import java.time.format.DateTimeParseException;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.kylin.common.KylinConfig;
+import org.apache.kylin.common.util.Pair;
+import org.apache.kylin.guava30.shaded.common.collect.Lists;
+import org.apache.kylin.guava30.shaded.common.collect.Maps;
+import org.apache.kylin.job.execution.AbstractExecutable;
+import org.apache.kylin.job.execution.ExecutableManager;
+import org.apache.kylin.job.execution.ExecutableState;
+import org.apache.kylin.job.execution.JobTypeEnum;
+import org.apache.kylin.job.util.JobContextUtil;
+import org.apache.kylin.metadata.model.NDataModel;
+import org.apache.kylin.metadata.model.NDataModelManager;
+import org.apache.kylin.metadata.model.PartitionDesc;
+import org.apache.kylin.metadata.project.NProjectManager;
+import org.apache.kylin.metadata.project.ProjectInstance;
+import org.apache.kylin.rest.service.ModelBuildService;
+import org.apache.kylin.rest.service.params.IncrementBuildSegmentParams;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.scheduling.TaskScheduler;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
+import org.springframework.stereotype.Component;
+
+import lombok.Getter;
+import lombok.val;
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+@Component
+public class AutoBuildSegmentScheduler {
+ private static final int THREAD_POOL_TASK_SCHEDULER_DEFAULT_POOL_SIZE = 20;
+ private static final DateTimeFormatter TIME_FORMATTER =
DateTimeFormatter.ofPattern("HH:mm:ss", Locale.ROOT);
+
+ @Autowired
+ @Qualifier("projectScheduler")
+ private TaskScheduler projectScheduler;
+
+ @Autowired
+ @Qualifier("modelBuildService")
+ private ModelBuildService modelBuildService;
+
+ @Getter
+ private final Map<String, Pair<String, ScheduledFuture<?>>> taskFutures =
Maps.newConcurrentMap();
+ @Getter
+ private final AtomicInteger schedulerModelCount = new AtomicInteger(0);
+
+ @Scheduled(cron = "*/30 * * * * ?")
+ public void schedulerAutoBuildSegment() {
+ val projectManager =
NProjectManager.getInstance(KylinConfig.readSystemKylinConfig());
+ reconcile(projectManager);
+ }
+
+ private void reconcile(NProjectManager projectManager) {
+ Map<String, String> expected = Maps.newHashMap();
+ for (ProjectInstance project : projectManager.listAllProjects()) {
+ val projectName = project.getName();
+ val modelManager =
NDataModelManager.getInstance(KylinConfig.readSystemKylinConfig(), projectName);
+ for (NDataModel model : modelManager.listAllModels()) {
+ val segmentConfig = model.getSegmentConfig();
+ if (segmentConfig == null ||
segmentConfig.getAutoSegmentBuild() == null
+ || !segmentConfig.getAutoSegmentBuild().isEnabled()) {
+ continue;
+ }
+ if (model.isStreaming() || model.isMultiPartitionModel()
+ ||
PartitionDesc.isEmptyPartitionDesc(model.getPartitionDesc())) {
+ continue;
+ }
+ val triggerTime =
segmentConfig.getAutoSegmentBuild().getTriggerTime();
+ if (StringUtils.isBlank(triggerTime)) {
+ continue;
+ }
+ val cron = toDailyCron(triggerTime);
+ if (cron == null) {
+ continue;
+ }
+ expected.put(buildTaskKey(projectName, model.getUuid()), cron);
+ }
+ }
+ for (val entry : expected.entrySet()) {
+ val key = entry.getKey();
+ val cron = entry.getValue();
+ val scheduled = taskFutures.get(key);
+ if (scheduled != null && StringUtils.equals(scheduled.getFirst(),
cron)) {
+ continue;
+ }
+ startCron(key, cron);
+ }
+ for (val key : Lists.newArrayList(taskFutures.keySet())) {
+ if (!expected.containsKey(key)) {
+ stopCron(key);
+ }
+ }
+ }
+
+ private void startCron(String key, String cron) {
+ stopCron(key);
+ checkSchedulerThreadPoolSize();
+ val scheduledFuture = projectScheduler.schedule(() -> submitJob(key),
triggerContext -> {
+ val trigger = new
org.springframework.scheduling.support.CronTrigger(cron);
+ return trigger.nextExecutionTime(triggerContext);
+ });
+ taskFutures.put(key, Pair.newPair(cron, scheduledFuture));
+ log.info("Auto build segment start cron, key: {}, cron: {}", key,
cron);
+ }
+
+ private void stopCron(String key) {
+ val scheduledFuturePair = taskFutures.get(key);
+ if (scheduledFuturePair != null) {
+ val future = scheduledFuturePair.getSecond();
+ if (future != null) {
+ future.cancel(true);
+ }
+ taskFutures.remove(key);
+ schedulerModelCount.decrementAndGet();
+ log.info("Auto build segment stop cron, key: {}", key);
+ }
+ }
+
+ private void checkSchedulerThreadPoolSize() {
+ val scheduler = (ThreadPoolTaskScheduler) projectScheduler;
+ val poolSize = scheduler.getPoolSize();
+ val modelCount = schedulerModelCount.incrementAndGet();
+ if (modelCount > poolSize) {
+ scheduler.setPoolSize(modelCount);
+ } else if (modelCount < THREAD_POOL_TASK_SCHEDULER_DEFAULT_POOL_SIZE
+ && poolSize > THREAD_POOL_TASK_SCHEDULER_DEFAULT_POOL_SIZE) {
+
scheduler.setPoolSize(THREAD_POOL_TASK_SCHEDULER_DEFAULT_POOL_SIZE);
+ }
+ }
+
+ private void submitJob(String key) {
+ if
(!JobContextUtil.getJobContext(KylinConfig.getInstanceFromEnv()).getJobScheduler().isMaster())
{
+ return;
+ }
+ val parts = key.split("/", 2);
+ if (parts.length != 2) {
+ return;
Review Comment:
add error log, identify the abnormal key
> Support built-in auto scheduled segment build with model-level config
> ---------------------------------------------------------------------
>
> Key: KYLIN-6087
> URL: https://issues.apache.org/jira/browse/KYLIN-6087
> Project: Kylin
> Issue Type: New Feature
> Components: Job Engine
> Affects Versions: 5.0.3
> Reporter: huangsheng
> Assignee: huangsheng
> Priority: Minor
>
> Currently, in kylin, incremental segment build job can only be triggered
> manually. I would like to add an automatic timed trigger for segment build
> job function because now we all rely on external timing tools (such as
> crontab) to call the kylin incremental build api at regular intervals to
> start segment construction. For example, the build job of yesterday's segment
> is triggered at 1 a.m. every day
> I want to directly incorporate this function into kylin. My idea is to add a
> new entry for scheduled scheduling in the model configuration pop-up box. It
> is necessary to configure the daily trigger time and the time range for
> building this segment. There are two parameters to configure for the time
> range. One is the logical date, such as yesterday
> The day before yesterday, etc., one is a specific time, for example, from
> 0:00 to 24:00. Then the back end schedules build tasks at regular intervals
> based on this configuration.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
