handle conflict

Author: mayinrain

linkis-computation-governance/linkis-entrance/src/main/scala/org/apache/linkis/entrance/conf/EntranceConfiguration.scala

@@ -295,10 +295,44 @@ object EntranceConfiguration {
       "01002,01003,13005,13006,13012"
     ).getValue
 
-  val AI_SQL_TEST_MODE =
-    CommonVars("linkis.entrance.aisql.test.mode", true).getValue
+  val AI_SQL_TEST_MODE: Boolean =
+    CommonVars[Boolean]("linkis.entrance.aisql.test.mode", true).getValue
+
+  val AI_SQL_ENABLED: Boolean =
+    CommonVars[Boolean]("linkis.ai.sql.enabled", true).getValue
+
+  val AI_SQL_DEFAULT_SPARK_ENGINE_TYPE: String =
+    CommonVars[String]("linkis.ai.sql.default.spark.engine.type", "spark-3.4.4").getValue
+
+  val AI_SQL_CREATORS: String =
+    CommonVars[String]("linkis.ai.sql.support.creators", "IDE").getValue
 
   val AI_SQL_KEY = "linkis.ai.sql"
 
   val RETRY_NUM_KEY = "linkis.ai.retry.num"
+
+  val SPARK_SHUFFLE_SERVICE_ENABLED: Boolean =
+    CommonVars[Boolean]("linkis.spark.shuffle.service.enabled", true).getValue
+
+  val SPARK_EXECUTOR_CORES: Int =
+    CommonVars[Int]("spark.executor.cores", 7).getValue
+
+  val SPARK_EXECUTOR_MEMORY: String =
+    CommonVars[String]("spark.executor.memory", "28G").getValue
+
+  val SPARK_EXECUTOR_INSTANCES: Int =
+    CommonVars[Int]("spark.executor.instances", 0).getValue
+
+  val SPARK_DYNAMIC_ALLOCATION_ENABLED: Boolean =
+    CommonVars[Boolean]("spark.dynamicAllocation.enabled", true).getValue
+
+  val SPARK_DYNAMIC_ALLOCATION_MIN_EXECUTORS: Int =
+    CommonVars[Int]("spark.dynamicAllocation.minExecutors", 0).getValue
+
+  val SPARK_DYNAMIC_ALLOCATION_MAX_EXECUTORS: Int =
+    CommonVars[Int]("spark.dynamicAllocation.maxExecutors", Integer.MAX_VALUE).getValue
+
+  val SPARK_DYNAMIC_ALLOCATION_ADDITIONAL_CONFS: String =
+    CommonVars[String]("spark.dynamicAllocation.additional.confs", "").getValue
+
 }

linkis-computation-governance/linkis-entrance/src/main/scala/org/apache/linkis/entrance/interceptor/impl/AISQLTransformInterceptor.scala

@@ -17,9 +17,10 @@
 
 package org.apache.linkis.entrance.interceptor.impl
 
+import org.apache.linkis.common.utils.{Logging, Utils}
 import org.apache.linkis.common.utils.CodeAndRunTypeUtils.LANGUAGE_TYPE_AI_SQL
-import org.apache.linkis.common.utils.Logging
 import org.apache.linkis.entrance.conf.EntranceConfiguration
+import org.apache.linkis.entrance.conf.EntranceConfiguration._
 import org.apache.linkis.entrance.interceptor.EntranceInterceptor
 import org.apache.linkis.governance.common.entity.job.{JobAiRequest, JobRequest}
 import org.apache.linkis.governance.common.protocol.job.JobAiReqInsert
@@ -29,6 +30,8 @@ import org.apache.linkis.manager.label.utils.LabelUtil
 import org.apache.linkis.protocol.utils.TaskUtils
 import org.apache.linkis.rpc.Sender
 
+import org.apache.commons.lang3.StringUtils
+
 import org.springframework.beans.BeanUtils
 
 import java.{lang, util}
@@ -39,18 +42,19 @@ import scala.collection.JavaConverters._
 class AISQLTransformInterceptor extends EntranceInterceptor with Logging {
 
   override def apply(jobRequest: JobRequest, logAppender: lang.StringBuilder): JobRequest = {
-    // TODO 修改为变量
-    val aiSqlEnable = true
-    // 转换为小写
-    val supportAISQLCreator = "IDE"
+    val aiSqlEnable: Boolean = AI_SQL_ENABLED
+    val supportAISQLCreator: String = AI_SQL_CREATORS.toLowerCase()
     val sqlLanguage: String = LANGUAGE_TYPE_AI_SQL
-    val sparkEngineType = "spark-3.4.4"
+    val sparkEngineType: String = AI_SQL_DEFAULT_SPARK_ENGINE_TYPE
     val labels: util.List[Label[_]] = jobRequest.getLabels
     val codeType: String = LabelUtil.getCodeType(labels)
     // engineType and creator have been verified in LabelCheckInterceptor.
     val userCreatorOpt: Option[Label[_]] = labels.asScala.find(_.isInstanceOf[UserCreatorLabel])
     val creator: String = userCreatorOpt.get.asInstanceOf[UserCreatorLabel].getCreator
     val engineTypeLabelOpt: Option[Label[_]] = labels.asScala.find(_.isInstanceOf[EngineTypeLabel])
+
+    val startMap: util.Map[String, AnyRef] = TaskUtils.getStartupMap(jobRequest.getParams)
+
     // aiSql change to spark
     var currentEngineType: String =
       engineTypeLabelOpt.get.asInstanceOf[EngineTypeLabel].getEngineType
@@ -59,20 +63,51 @@ class AISQLTransformInterceptor extends EntranceInterceptor with Logging {
           .equals(codeType) && supportAISQLCreator.contains(creator.toLowerCase())
     ) {
       engineTypeLabelOpt.get.asInstanceOf[EngineTypeLabel].setEngineType(sparkEngineType)
-      // TODO 添加 aisql 标识
+
+      startMap.put(AI_SQL_KEY, "true")
+
       currentEngineType = sparkEngineType
 
-      // TODO 将转换后的数据保存到数据库
       persist(jobRequest);
 
     }
     // 开启 spark 动态资源规划, spark3.4.4
     if (sparkEngineType.equals(currentEngineType)) {
       logger.info("spark3 add dynamic resource.")
-      val startMap: util.Map[String, AnyRef] = TaskUtils.getStartupMap(jobRequest.getParams)
+
       // add spark dynamic resource planning
-      // TODO
-      startMap.put("", "")
+      startMap.put(
+        "spark.shuffle.service.enabled",
+        SPARK_SHUFFLE_SERVICE_ENABLED.asInstanceOf[AnyRef]
+      )
+      startMap.put(
+        "spark.dynamicAllocation.enabled",
+        SPARK_DYNAMIC_ALLOCATION_ENABLED.asInstanceOf[AnyRef]
+      )
+      startMap.put(
+        "spark.dynamicAllocation.minExecutors",
+        SPARK_DYNAMIC_ALLOCATION_MIN_EXECUTORS.asInstanceOf[AnyRef]
+      )
+      startMap.put(
+        "spark.dynamicAllocation.maxExecutors",
+        SPARK_DYNAMIC_ALLOCATION_MAX_EXECUTORS.asInstanceOf[AnyRef]
+      )
+      startMap.put("spark.executor.cores", SPARK_EXECUTOR_CORES.asInstanceOf[AnyRef])
+      startMap.put("spark.executor.memory", SPARK_EXECUTOR_MEMORY.asInstanceOf[AnyRef])
+      startMap.put("spark.executor.instances", SPARK_EXECUTOR_INSTANCES.asInstanceOf[AnyRef])
+
+      Utils.tryAndWarn {
+        val extraConfs: String = SPARK_DYNAMIC_ALLOCATION_ADDITIONAL_CONFS
+        if (StringUtils.isNotBlank(extraConfs)) {
+          val confs: Array[String] = extraConfs.split(",")
+          for (conf <- confs) {
+            val confKey: String = conf.split("=")(0)
+            val confValue: String = conf.split("=")(1)
+            startMap.put(confKey, confValue)
+          }
+        }
+      }
+
     }
     jobRequest
   }