You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@linkis.apache.org by ca...@apache.org on 2022/04/14 11:16:24 UTC
[incubator-linkis] branch dev-1.1.2 updated: Computation ec asynchronous execution supports setting the number of concurrent job groups (#1971)
This is an automated email from the ASF dual-hosted git repository.
casion pushed a commit to branch dev-1.1.2
in repository https://gitbox.apache.org/repos/asf/incubator-linkis.git
The following commit(s) were added to refs/heads/dev-1.1.2 by this push:
new 45d71d60d Computation ec asynchronous execution supports setting the number of concurrent job groups (#1971)
45d71d60d is described below
commit 45d71d60d56416e83f4cab2ae44372af0f420b7b
Author: peacewong <wp...@gmail.com>
AuthorDate: Thu Apr 14 19:16:17 2022 +0800
Computation ec asynchronous execution supports setting the number of concurrent job groups (#1971)
* update name to job tags
* Computation ec asynchronous execution supports setting the number of concurrent job groups #1970
* add ec task id to task log
---
.../engineconn/computation/executor/async/AsyncExecuteContext.scala | 6 ++++++
.../computation/executor/conf/ComputationExecutorConf.scala | 2 ++
.../linkis/engineplugin/hive/executor/HiveEngineConnExecutor.scala | 4 ++--
.../orchestrator/computation/physical/CodeLogicalUnitExecTask.scala | 3 ++-
4 files changed, 12 insertions(+), 3 deletions(-)
diff --git a/linkis-computation-governance/linkis-engineconn/linkis-computation-engineconn/src/main/scala/org/apache/linkis/engineconn/computation/executor/async/AsyncExecuteContext.scala b/linkis-computation-governance/linkis-engineconn/linkis-computation-engineconn/src/main/scala/org/apache/linkis/engineconn/computation/executor/async/AsyncExecuteContext.scala
index cf932b5fb..896cb8138 100644
--- a/linkis-computation-governance/linkis-engineconn/linkis-computation-engineconn/src/main/scala/org/apache/linkis/engineconn/computation/executor/async/AsyncExecuteContext.scala
+++ b/linkis-computation-governance/linkis-engineconn/linkis-computation-engineconn/src/main/scala/org/apache/linkis/engineconn/computation/executor/async/AsyncExecuteContext.scala
@@ -18,6 +18,7 @@
package org.apache.linkis.engineconn.computation.executor.async
import org.apache.linkis.engineconn.computation.executor.conf.ComputationExecutorConf
+import org.apache.linkis.scheduler.queue.fifoqueue.FIFOGroupFactory
import org.apache.linkis.scheduler.queue.parallelqueue.{ParallelScheduler, ParallelSchedulerContextImpl}
import org.apache.linkis.scheduler.{Scheduler, SchedulerContext}
@@ -44,6 +45,11 @@ class AsyncExecuteContextImpl extends AsyncExecuteContext {
private def createSchedulerContext(executor: AsyncConcurrentComputationExecutor): SchedulerContext = {
val parallelSchedulerContext = new ParallelSchedulerContextImpl(ComputationExecutorConf.ASYNC_EXECUTE_MAX_PARALLELISM.getValue)
parallelSchedulerContext.setExecutorManager(new AsyncExecutorManager(executor))
+ parallelSchedulerContext.getOrCreateGroupFactory match {
+ case groupFactory: FIFOGroupFactory =>
+ groupFactory.setDefaultMaxRunningJobs(ComputationExecutorConf.ASYNC_SCHEDULER_MAX_RUNNING_JOBS)
+ case _ =>
+ }
parallelSchedulerContext
}
diff --git a/linkis-computation-governance/linkis-engineconn/linkis-computation-engineconn/src/main/scala/org/apache/linkis/engineconn/computation/executor/conf/ComputationExecutorConf.scala b/linkis-computation-governance/linkis-engineconn/linkis-computation-engineconn/src/main/scala/org/apache/linkis/engineconn/computation/executor/conf/ComputationExecutorConf.scala
index a1ebcf1e0..2b5ba62c5 100644
--- a/linkis-computation-governance/linkis-engineconn/linkis-computation-engineconn/src/main/scala/org/apache/linkis/engineconn/computation/executor/conf/ComputationExecutorConf.scala
+++ b/linkis-computation-governance/linkis-engineconn/linkis-computation-engineconn/src/main/scala/org/apache/linkis/engineconn/computation/executor/conf/ComputationExecutorConf.scala
@@ -43,6 +43,8 @@ object ComputationExecutorConf {
val ASYNC_EXECUTE_MAX_PARALLELISM = CommonVars("wds.linkis.engineconn.max.parallelism", 300)
+ val ASYNC_SCHEDULER_MAX_RUNNING_JOBS = CommonVars("wds.linkis.engineconn.async.group.max.running", 10).getValue
+
val DEFAULT_COMPUTATION_EXECUTORMANAGER_CLAZZ = CommonVars("wds.linkis.default.computation.executormanager.clazz", "org.apache.linkis.engineconn.computation.executor.creation.ComputationExecutorManagerImpl")
diff --git a/linkis-engineconn-plugins/engineconn-plugins/hive/src/main/scala/org/apache/linkis/engineplugin/hive/executor/HiveEngineConnExecutor.scala b/linkis-engineconn-plugins/engineconn-plugins/hive/src/main/scala/org/apache/linkis/engineplugin/hive/executor/HiveEngineConnExecutor.scala
index df2840942..68875f9ca 100644
--- a/linkis-engineconn-plugins/engineconn-plugins/hive/src/main/scala/org/apache/linkis/engineplugin/hive/executor/HiveEngineConnExecutor.scala
+++ b/linkis-engineconn-plugins/engineconn-plugins/hive/src/main/scala/org/apache/linkis/engineplugin/hive/executor/HiveEngineConnExecutor.scala
@@ -123,8 +123,8 @@ class HiveEngineConnExecutor(id: Int,
LOG.info(s"hive client begins to run hql code:\n ${realCode.trim}")
val jobId = JobUtils.getJobIdFromMap(engineExecutorContext.getProperties)
if (StringUtils.isNotBlank(jobId)) {
- LOG.info(s"set mapred.job.name=LINKIS_$jobId")
- hiveConf.set("mapred.job.name", s"LINKIS_$jobId")
+ LOG.info(s"set mapreduce.job.tags=LINKIS_$jobId")
+ hiveConf.set("mapreduce.job.tags", s"LINKIS_$jobId")
}
if (realCode.trim.length > 500) {
engineExecutorContext.appendStdout(s"$getId >> ${realCode.trim.substring(0, 500)} ...")
diff --git a/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/physical/CodeLogicalUnitExecTask.scala b/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/physical/CodeLogicalUnitExecTask.scala
index 9c3804870..f31f89d73 100644
--- a/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/physical/CodeLogicalUnitExecTask.scala
+++ b/linkis-orchestrator/linkis-computation-orchestrator/src/main/scala/org/apache/linkis/orchestrator/computation/physical/CodeLogicalUnitExecTask.scala
@@ -18,7 +18,6 @@
package org.apache.linkis.orchestrator.computation.physical
import java.util.concurrent.TimeUnit
-
import org.apache.linkis.common.exception.{ErrorException, LinkisRetryException, WarnException}
import org.apache.linkis.common.log.LogUtils
import org.apache.linkis.common.utils.{Logging, Utils}
@@ -40,6 +39,7 @@ import org.apache.linkis.orchestrator.strategy.{ResultSetExecTask, StatusInfoExe
import org.apache.linkis.orchestrator.utils.OrchestratorIDCreator
import org.apache.linkis.scheduler.executer.{ErrorExecuteResponse, SubmitResponse}
import org.apache.commons.lang.StringUtils
+import org.apache.linkis.orchestrator.listener.task.TaskLogEvent
import scala.concurrent.duration.Duration
import scala.collection.convert.decorateAsScala._
@@ -89,6 +89,7 @@ class CodeLogicalUnitExecTask (parents: Array[ExecTask], children: Array[ExecTas
//封装engineConnExecId信息
codeExecutor.setEngineConnTaskId(engineConnExecId)
codeExecTaskExecutorManager.addEngineConnTaskID(codeExecutor)
+ getPhysicalContext.pushLog(TaskLogEvent(this, LogUtils.generateInfo(s"Task submit to ec: ${codeExecutor.getEngineConnExecutor.getServiceInstance} get engineConnExecId is: ${engineConnExecId}")))
new AsyncTaskResponse {
override def notifyMe(listener: NotifyListener): Unit = null
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@linkis.apache.org
For additional commands, e-mail: commits-help@linkis.apache.org