diff --git a/docs/configuration/spark.md b/docs/configuration/spark.md
index f40c76b43d..ed070e2ac4 100644
--- a/docs/configuration/spark.md
+++ b/docs/configuration/spark.md
@@ -29,7 +29,10 @@
 |spark|wds.linkis.spark.engine.scala.replace_package_header.enable| true |spark.engine.scala.replace_package_header.enable|
 
 Use spark yarn cluster mode,need to set label "engingeConnRuntimeMode": "yarnCluster",and need to upload the dependence of the spark to 'linkis.spark.yarn.cluster.jar'(the default value is 'hdfs:///spark/cluster')
-spark dependencies include jars and configuration files，For example: '/appcom/Install/linkis/lib/linkis-engineconn-plugins/spark/dist/3.2.1/lib/*.jar','/appcom/Install/linkis/conf/*''
+spark dependencies include jars and configuration files，For example: '/appcom/Install/linkis/lib/linkis-engineconn-plugins/spark/dist/3.2.1/lib/*.jar','/appcom/Install/linkis/conf/*'
+
+Precautions for using yarnCluster:
+Eureka url if 127.0.0.1 should be changed to the real host, such as "127.0.0.1:20303/eureka/" should be changed to "wds001:20303/eureka/"
 
 The spark-excel package may cause class conflicts,need to download separately,put it in spark lib
 wget https://repo1.maven.org/maven2/com/crealytics/spark-excel-2.12.17-3.2.2_2.12/3.2.2_0.18.1/spark-excel-2.12.17-3.2.2_2.12-3.2.2_0.18.1.jar
diff --git a/linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/interactor/job/interactive/InteractiveJob.java b/linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/interactor/job/interactive/InteractiveJob.java
index 9e1733ccf1..a097b1b25b 100644
--- a/linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/interactor/job/interactive/InteractiveJob.java
+++ b/linkis-computation-governance/linkis-client/linkis-cli/src/main/java/org/apache/linkis/cli/application/interactor/job/interactive/InteractiveJob.java
@@ -41,6 +41,7 @@
 import org.apache.commons.lang3.exception.ExceptionUtils;
 
 import java.util.HashMap;
+import java.util.Map;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -116,7 +117,7 @@ public JobResult run() {
     // async job, return
     if (isAsync) {
       return new InteractiveJobResult(
-          submitResult.getJobStatus().isJobSubmitted(),
+          jobInfoResult.getJobStatus().isJobSubmitted(),
           "Async Submission Success",
           new HashMap<>());
     }
@@ -171,7 +172,14 @@ private JobResult getResult(
               "Job status is not success but \'"
                   + jobInfoResult.getJobStatus()
                   + "\'. Will not try to retrieve any Result");
-      return new InteractiveJobResult(false, "Execute Error!!!", new HashMap<>());
+      Map<String, String> extraMap = new HashMap<>();
+      if (jobInfoResult.getErrCode() != null) {
+        extraMap.put("errorCode", String.valueOf(jobInfoResult.getErrCode()));
+      }
+      if (StringUtils.isNotBlank(jobInfoResult.getErrDesc())) {
+        extraMap.put("errorDesc", jobInfoResult.getErrDesc());
+      }
+      return new InteractiveJobResult(false, "Execute Error!!!", extraMap);
     }
     InteractiveJobResult result =
         new InteractiveJobResult(true, "Execute Success!!!", new HashMap<>());
@@ -246,6 +254,9 @@ public void onDestroy() {
       logger.warn("Failed to kill job username or jobId is blank");
       return;
     }
+    if (isAsync) {
+      return;
+    }
     try {
       new JobKiller(oper).doKill(username, jobId);
     } catch (Exception e) {
diff --git a/linkis-computation-governance/linkis-engineconn/linkis-engineconn-executor/accessible-executor/src/main/scala/org/apache/linkis/engineconn/callback/hook/CallbackEngineConnHook.scala b/linkis-computation-governance/linkis-engineconn/linkis-engineconn-executor/accessible-executor/src/main/scala/org/apache/linkis/engineconn/callback/hook/CallbackEngineConnHook.scala
index adcbb1a695..07cfa51d0a 100644
--- a/linkis-computation-governance/linkis-engineconn/linkis-engineconn-executor/accessible-executor/src/main/scala/org/apache/linkis/engineconn/callback/hook/CallbackEngineConnHook.scala
+++ b/linkis-computation-governance/linkis-engineconn/linkis-engineconn-executor/accessible-executor/src/main/scala/org/apache/linkis/engineconn/callback/hook/CallbackEngineConnHook.scala
@@ -61,15 +61,16 @@ class CallbackEngineConnHook extends EngineConnHook with Logging {
     newMap.put("spring.mvc.servlet.path", ServerConfiguration.BDP_SERVER_RESTFUL_URI.getValue)
     DataWorkCloudApplication.main(DWCArgumentsParser.formatSpringOptions(newMap.toMap))
 
-    val engineConnIdentifierCallback = new EngineConnIdentifierCallback()
-    Utils.tryAndError(engineConnIdentifierCallback.callback())
     logger.info("<--------------------SpringBoot App init succeed-------------------->")
   }
 
   override def beforeExecutionExecute(
       engineCreationContext: EngineCreationContext,
       engineConn: EngineConn
-  ): Unit = {}
+  ): Unit = {
+    val engineConnIdentifierCallback = new EngineConnIdentifierCallback()
+    Utils.tryAndError(engineConnIdentifierCallback.callback())
+  }
 
   override def afterExecutionExecute(
       engineCreationContext: EngineCreationContext,
diff --git a/linkis-computation-governance/linkis-jdbc-driver/src/main/scala/org/apache/linkis/ujes/jdbc/LinkisSQLConnection.scala b/linkis-computation-governance/linkis-jdbc-driver/src/main/scala/org/apache/linkis/ujes/jdbc/LinkisSQLConnection.scala
index b800698766..e111615cee 100644
--- a/linkis-computation-governance/linkis-jdbc-driver/src/main/scala/org/apache/linkis/ujes/jdbc/LinkisSQLConnection.scala
+++ b/linkis-computation-governance/linkis-jdbc-driver/src/main/scala/org/apache/linkis/ujes/jdbc/LinkisSQLConnection.scala
@@ -431,6 +431,7 @@ class LinkisSQLConnection(private[jdbc] val ujesClient: UJESClient, props: Prope
       case EngineType.HIVE => RunType.HIVE
       case EngineType.TRINO => RunType.TRINO_SQL
       case EngineType.PRESTO => RunType.PRESTO_SQL
+      case EngineType.NEBULA => RunType.NEBULA_SQL
       case EngineType.ELASTICSEARCH => RunType.ES_SQL
       case EngineType.JDBC => RunType.JDBC
       case EngineType.PYTHON => RunType.SHELL
diff --git a/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/am/restful/EMRestfulApi.java b/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/am/restful/EMRestfulApi.java
index c0da8cde24..4d5cb480d3 100644
--- a/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/am/restful/EMRestfulApi.java
+++ b/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/am/restful/EMRestfulApi.java
@@ -359,9 +359,11 @@ public Message executeECMOperation(HttpServletRequest req, @RequestBody JsonNode
           "Fail to process the operation parameters, cased by "
               + ExceptionUtils.getRootCauseMessage(e));
     }
-    String engineConnInstance = (String) parameters.get("engineConnInstance");
+
     return executeECMOperation(
-        ecmNode, engineConnInstance, new ECMOperateRequest(userName, parameters));
+        ecmNode,
+        parameters.getOrDefault("engineConnInstance", "").toString(),
+        new ECMOperateRequest(userName, parameters));
   }
 
   @ApiOperation(value = "openEngineLog", notes = "open Engine log", response = Message.class)
diff --git a/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/am/selector/rule/ResourceNodeSelectRule.java b/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/am/selector/rule/ResourceNodeSelectRule.java
index 200794671a..ad259ec30c 100644
--- a/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/am/selector/rule/ResourceNodeSelectRule.java
+++ b/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/am/selector/rule/ResourceNodeSelectRule.java
@@ -53,10 +53,10 @@ private Comparator<Node> sortByResource() {
           RMNode nodeBRm = (RMNode) nodeB;
           if (nodeARm.getNodeResource() == null
               || nodeARm.getNodeResource().getLeftResource() == null) {
-            return -1;
+            return 1;
           } else if (nodeBRm.getNodeResource() == null
               || nodeBRm.getNodeResource().getLeftResource() == null) {
-            return 1;
+            return -1;
           } else {
             if (nodeARm
                 .getNodeResource()
@@ -67,9 +67,13 @@ private Comparator<Node> sortByResource() {
                 .getNodeResource()
                 .getLeftResource()
                 .moreThan(nodeBRm.getNodeResource().getLeftResource())) {
-              return 1;
-            } else {
               return -1;
+            } else {
+              // 从大到小排序 (Sort from large to small)
+              return -(nodeARm
+                  .getNodeResource()
+                  .getLeftResource()
+                  .compare(nodeBRm.getNodeResource().getLeftResource()));
             }
           }
         } catch (Throwable t) {
@@ -93,10 +97,10 @@ private Comparator<Node> sortByResourceRate() {
           RMNode nodeBRm = (RMNode) nodeB;
           if (nodeARm.getNodeResource() == null
               || nodeARm.getNodeResource().getLeftResource() == null) {
-            return -1;
+            return 1;
           } else if (nodeBRm.getNodeResource() == null
               || nodeBRm.getNodeResource().getLeftResource() == null) {
-            return 1;
+            return -1;
           } else {
             float aRate =
                 ResourceUtils.getLoadInstanceResourceRate(
@@ -106,7 +110,7 @@ private Comparator<Node> sortByResourceRate() {
                 ResourceUtils.getLoadInstanceResourceRate(
                     nodeBRm.getNodeResource().getLeftResource(),
                     nodeBRm.getNodeResource().getMaxResource());
-            return Float.compare(aRate, bRate);
+            return -Float.compare(aRate, bRate);
           }
         } catch (Throwable t) {
           logger.warn("Failed to Compare resource " + t.getMessage());
diff --git a/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/am/selector/rule/ScoreNodeSelectRule.java b/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/am/selector/rule/ScoreNodeSelectRule.java
index 43bf789d09..dae02bec5b 100644
--- a/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/am/selector/rule/ScoreNodeSelectRule.java
+++ b/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/am/selector/rule/ScoreNodeSelectRule.java
@@ -55,13 +55,13 @@ private Comparator<Node> sortByScore() {
         ScoreServiceInstance instanceB = (ScoreServiceInstance) nodeB;
         try {
           if (instanceA.getScore() > instanceB.getScore()) {
-            return 1;
+            return -1;
           }
         } catch (Exception e) {
           logger.warn("Failed to Compare resource ", e);
           return -1;
         }
-        return -1;
+        return 1;
       } else {
         return -1;
       }
diff --git a/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/am/service/impl/DefaultEngineConnPidCallbackService.java b/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/am/service/impl/DefaultEngineConnPidCallbackService.java
index 4acfb70f91..5fbbb7c32a 100644
--- a/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/am/service/impl/DefaultEngineConnPidCallbackService.java
+++ b/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/am/service/impl/DefaultEngineConnPidCallbackService.java
@@ -54,7 +54,8 @@ public void dealPid(ResponseEngineConnPid protocol) {
         protocol.pid(),
         protocol.ticketId());
 
-    EngineNode engineNode = defaultEngineNodeManager.getEngineNode(protocol.serviceInstance());
+    EngineNode engineNode =
+        defaultEngineNodeManager.getEngineNodeInfoByTicketId(protocol.ticketId());
     if (engineNode == null) {
       logger.error(
           "DefaultEngineConnPidCallbackService dealPid failed, engineNode is null, serviceInstance:{}",
@@ -63,13 +64,12 @@ public void dealPid(ResponseEngineConnPid protocol) {
     }
 
     engineNode.setIdentifier(protocol.pid());
-
+    ServiceInstance oldServiceInstance = engineNode.getServiceInstance();
     if (engineNode.getMark().equals(AMConstant.CLUSTER_PROCESS_MARK)) {
       ServiceInstance serviceInstance = protocol.serviceInstance();
       engineNode.setServiceInstance(serviceInstance);
-      getEngineNodeManager().updateEngineNode(serviceInstance, engineNode);
-      nodeLabelService.labelsFromInstanceToNewInstance(
-          engineNode.getServiceInstance(), serviceInstance);
+      getEngineNodeManager().updateEngineNode(oldServiceInstance, engineNode);
+      nodeLabelService.labelsFromInstanceToNewInstance(oldServiceInstance, serviceInstance);
     }
     defaultEngineNodeManager.updateEngine(engineNode);
   }
diff --git a/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/label/conf/LabelManagerConf.java b/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/label/conf/LabelManagerConf.java
index f436254911..9aa5ff797f 100644
--- a/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/label/conf/LabelManagerConf.java
+++ b/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/label/conf/LabelManagerConf.java
@@ -23,4 +23,7 @@ public class LabelManagerConf {
 
   public static final String LONG_LIVED_LABEL =
       CommonVars.apply("wds.linkis.label.node.long.lived.label.keys", "tenant").getValue();
+
+  public static final boolean COMBINED_WITHOUT_YARN_DEFAULT =
+      CommonVars.apply("linkis.combined.without.yarn.default", true).getValue();
 }
diff --git a/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/rm/domain/RMLabelContainer.java b/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/rm/domain/RMLabelContainer.java
index 5bda339194..9d3140267b 100644
--- a/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/rm/domain/RMLabelContainer.java
+++ b/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/rm/domain/RMLabelContainer.java
@@ -18,10 +18,13 @@
 package org.apache.linkis.manager.rm.domain;
 
 import org.apache.linkis.governance.common.conf.GovernanceCommonConf;
+import org.apache.linkis.manager.common.conf.RMConfiguration;
 import org.apache.linkis.manager.label.builder.CombinedLabelBuilder;
+import org.apache.linkis.manager.label.conf.LabelManagerConf;
 import org.apache.linkis.manager.label.entity.CombinedLabel;
 import org.apache.linkis.manager.label.entity.Label;
 import org.apache.linkis.manager.label.entity.ResourceLabel;
+import org.apache.linkis.manager.label.entity.cluster.ClusterLabel;
 import org.apache.linkis.manager.label.entity.em.EMInstanceLabel;
 import org.apache.linkis.manager.label.entity.engine.EngineInstanceLabel;
 import org.apache.linkis.manager.label.entity.engine.EngineTypeLabel;
@@ -49,7 +52,8 @@ public class RMLabelContainer {
   private EngineTypeLabel engineTypeLabel;
   private UserCreatorLabel userCreatorLabel;
   private EngineInstanceLabel engineInstanceLabel;
-  private CombinedLabel combinedUserCreatorEngineTypeLabel;
+  private ClusterLabel clusterLabel;
+  private CombinedLabel combinedResourceLabel;
   private Label currentLabel;
 
   public RMLabelContainer(List<Label<?>> labels) {
@@ -57,14 +61,16 @@ public RMLabelContainer(List<Label<?>> labels) {
     this.lockedLabels = Lists.newArrayList();
     try {
       if (getUserCreatorLabel() != null && getEngineTypeLabel() != null) {
-        this.combinedUserCreatorEngineTypeLabel =
-            (CombinedLabel)
-                combinedLabelBuilder.build(
-                    "", Lists.newArrayList(getUserCreatorLabel(), getEngineTypeLabel()));
-        this.labels.add(combinedUserCreatorEngineTypeLabel);
+        List<Label> combinedLabel = Lists.newArrayList(getUserCreatorLabel(), getEngineTypeLabel());
+        ClusterLabel clusterLabel = getClusterLabel();
+        if (shouldCombinedClusterLabel(clusterLabel)) {
+          combinedLabel.add(clusterLabel);
+        }
+        this.combinedResourceLabel = (CombinedLabel) combinedLabelBuilder.build("", combinedLabel);
+        this.labels.add(combinedResourceLabel);
       }
     } catch (Exception e) {
-      logger.warn("failed to get combinedUserCreatorEngineTypeLabel", e);
+      logger.warn("failed to get combinedResourceLabel", e);
     }
     this.labels = LabelUtils.distinctLabel(this.labels, labels);
   }
@@ -156,8 +162,31 @@ public EngineInstanceLabel getEngineInstanceLabel() {
     return null;
   }
 
-  public CombinedLabel getCombinedUserCreatorEngineTypeLabel() {
-    return combinedUserCreatorEngineTypeLabel;
+  public ClusterLabel getClusterLabel() {
+    if (clusterLabel == null) {
+      for (Label label : labels) {
+        if (label instanceof ClusterLabel) {
+          return (ClusterLabel) label;
+        }
+      }
+    } else {
+      return clusterLabel;
+    }
+    logger.warn("ClusterLabel not found");
+    return null;
+  }
+
+  private boolean shouldCombinedClusterLabel(ClusterLabel clusterLabel) {
+    return !(clusterLabel == null
+        || (LabelManagerConf.COMBINED_WITHOUT_YARN_DEFAULT
+            && clusterLabel
+                .getClusterName()
+                .equals(RMConfiguration.DEFAULT_YARN_CLUSTER_NAME.getValue())
+            && clusterLabel.getClusterType().equals(RMConfiguration.DEFAULT_YARN_TYPE.getValue())));
+  }
+
+  public CombinedLabel getCombinedResourceLabel() {
+    return combinedResourceLabel;
   }
 
   public Label getCurrentLabel() {
@@ -195,6 +224,8 @@ public String toString() {
         + userCreatorLabel
         + ", engineInstanceLabel="
         + engineInstanceLabel
+        + ", clusterLabel="
+        + clusterLabel
         + ", currentLabel="
         + currentLabel
         + '}';
diff --git a/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/rm/external/kubernetes/KubernetesResourceRequester.java b/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/rm/external/kubernetes/KubernetesResourceRequester.java
index 3bc7004e3f..c886b622ab 100644
--- a/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/rm/external/kubernetes/KubernetesResourceRequester.java
+++ b/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/rm/external/kubernetes/KubernetesResourceRequester.java
@@ -24,8 +24,11 @@
 import org.apache.linkis.manager.rm.external.request.ExternalResourceRequester;
 
 import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.io.FileUtils;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.commons.lang3.tuple.Pair;
 
+import java.io.File;
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
 
@@ -33,8 +36,10 @@
 import io.fabric8.kubernetes.api.model.Quantity;
 import io.fabric8.kubernetes.api.model.ResourceQuota;
 import io.fabric8.kubernetes.api.model.metrics.v1beta1.NodeMetrics;
+import io.fabric8.kubernetes.client.Config;
 import io.fabric8.kubernetes.client.ConfigBuilder;
 import io.fabric8.kubernetes.client.DefaultKubernetesClient;
+import io.fabric8.kubernetes.client.KubernetesClientException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -45,7 +50,7 @@ public class KubernetesResourceRequester implements ExternalResourceRequester {
   @Override
   public NodeResource requestResourceInfo(
       ExternalResourceIdentifier identifier, ExternalResourceProvider provider) {
-    String k8sMasterUrl = (String) provider.getConfigMap().get("k8sMasterUrl");
+    String k8sMasterUrl = getK8sMasterUrl(provider);
     DefaultKubernetesClient client = clientMap.get(k8sMasterUrl);
     if (client == null) {
       constructKubernetesClient(provider);
@@ -154,8 +159,7 @@ public ResourceType getResourceType() {
   @Override
   public Boolean reloadExternalResourceAddress(ExternalResourceProvider provider) {
     if (null != provider) {
-      DefaultKubernetesClient client =
-          clientMap.get((String) provider.getConfigMap().get("k8sMasterUrl"));
+      DefaultKubernetesClient client = clientMap.get(getK8sMasterUrl(provider));
       if (client != null) {
         client.close();
       }
@@ -164,19 +168,42 @@ public Boolean reloadExternalResourceAddress(ExternalResourceProvider provider)
     return true;
   }
 
+  private String getK8sMasterUrl(ExternalResourceProvider provider) {
+    Map<String, Object> configMap = provider.getConfigMap();
+    String k8sMasterUrl = (String) configMap.get("k8sMasterUrl");
+    if (StringUtils.isBlank(k8sMasterUrl)) {
+      throw new IllegalArgumentException("k8sMasterUrl is empty, please check the configuration.");
+    }
+    return k8sMasterUrl;
+  }
+
   private void constructKubernetesClient(ExternalResourceProvider provider) {
-    String k8sMasterUrl = (String) provider.getConfigMap().get("k8sMasterUrl");
-    String k8sClientCertData = (String) provider.getConfigMap().get("k8sClientCertData");
-    String k8sClientKeyData = (String) provider.getConfigMap().get("k8sClientKeyData");
-    String k8sCaCertData = (String) provider.getConfigMap().get("k8sCaCertData");
-    DefaultKubernetesClient client =
-        new DefaultKubernetesClient(
-            new ConfigBuilder()
-                .withMasterUrl(k8sMasterUrl)
-                .withClientCertData(k8sClientCertData)
-                .withClientKeyData(k8sClientKeyData)
-                .withCaCertData(k8sCaCertData)
-                .build());
+    DefaultKubernetesClient client;
+    Map<String, Object> configMap = provider.getConfigMap();
+    String k8sMasterUrl = getK8sMasterUrl(provider);
+    try {
+      String k8sConfig = (String) configMap.get("k8sConfig");
+      if (StringUtils.isNotBlank(k8sConfig)) {
+        Config kubeConfig =
+            Config.fromKubeconfig(
+                null, FileUtils.readFileToString(new File(k8sConfig), "UTF-8"), null);
+        client = new DefaultKubernetesClient(kubeConfig);
+      } else {
+        String k8sClientCertData = (String) configMap.get("k8sClientCertData");
+        String k8sClientKeyData = (String) configMap.get("k8sClientKeyData");
+        String k8sCaCertData = (String) configMap.get("k8sCaCertData");
+        client =
+            new DefaultKubernetesClient(
+                new ConfigBuilder()
+                    .withMasterUrl(k8sMasterUrl)
+                    .withClientCertData(k8sClientCertData)
+                    .withClientKeyData(k8sClientKeyData)
+                    .withCaCertData(k8sCaCertData)
+                    .build());
+      }
+    } catch (Exception e) {
+      throw new KubernetesClientException("Fail to build k8s client. ", e);
+    }
     clientMap.put(k8sMasterUrl, client);
   }
 }
diff --git a/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/rm/message/RMMessageService.java b/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/rm/message/RMMessageService.java
index ff58f933b6..88985bf52e 100644
--- a/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/rm/message/RMMessageService.java
+++ b/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/rm/message/RMMessageService.java
@@ -53,9 +53,7 @@ public void dealWithResourceUsedProtocol(ResourceUsedProtocol resourceUsedProtoc
     } catch (Exception e) {
       RMLabelContainer nodeLabels = new RMLabelContainer(labels);
       String value =
-          Optional.of(nodeLabels.getCombinedUserCreatorEngineTypeLabel())
-              .map(Object::toString)
-              .orElse("");
+          Optional.of(nodeLabels.getCombinedResourceLabel()).map(Object::toString).orElse("");
       logger.warn(
           String.format(
               "usedResource failed, request from:%s, request engine: %s, ",
diff --git a/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/rm/service/RequestResourceService.java b/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/rm/service/RequestResourceService.java
index 4c1237cc45..c88b39e52a 100644
--- a/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/rm/service/RequestResourceService.java
+++ b/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/rm/service/RequestResourceService.java
@@ -59,9 +59,7 @@ public boolean canRequest(RMLabelContainer labelContainer, NodeResource resource
     NodeResource labelResource =
         labelResourceService.getLabelResource(labelContainer.getCurrentLabel());
     Resource requestResource = resource.getMinResource();
-    if (labelContainer
-        .getCombinedUserCreatorEngineTypeLabel()
-        .equals(labelContainer.getCurrentLabel())) {
+    if (labelContainer.getCombinedResourceLabel().equals(labelContainer.getCurrentLabel())) {
       if (labelResource == null) {
         labelResource = new CommonNodeResource();
         labelResource.setResourceType(resource.getResourceType());
@@ -92,7 +90,7 @@ public boolean canRequest(RMLabelContainer labelContainer, NodeResource resource
       labelResourceService.setLabelResource(
           labelContainer.getCurrentLabel(),
           labelResource,
-          labelContainer.getCombinedUserCreatorEngineTypeLabel().getStringValue());
+          labelContainer.getCombinedResourceLabel().getStringValue());
       logger.debug(
           labelContainer.getCurrentLabel()
               + " to request ["
diff --git a/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/rm/service/impl/DefaultResourceManager.java b/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/rm/service/impl/DefaultResourceManager.java
index c402701450..672963e306 100644
--- a/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/rm/service/impl/DefaultResourceManager.java
+++ b/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/rm/service/impl/DefaultResourceManager.java
@@ -247,8 +247,7 @@ public ResultResource requestResource(List<Label<?>> labels, NodeResource resour
 
     List<PersistenceLock> persistenceLocks = new ArrayList<>();
     EMInstanceLabel emInstanceLabel = labelContainer.getEMInstanceLabel();
-    CombinedLabel userCreatorEngineTypeLabel =
-        labelContainer.getCombinedUserCreatorEngineTypeLabel();
+    CombinedLabel combinedLabel = labelContainer.getCombinedResourceLabel();
 
     try {
       // check ecm resource if not enough return
@@ -266,14 +265,12 @@ public ResultResource requestResource(List<Label<?>> labels, NodeResource resour
 
       // lock userCreatorEngineTypeLabel
       persistenceLocks.add(
-          tryLockOneLabel(
-              userCreatorEngineTypeLabel, wait, labelContainer.getUserCreatorLabel().getUser()));
+          tryLockOneLabel(combinedLabel, wait, labelContainer.getUserCreatorLabel().getUser()));
       try {
-        labelContainer.setCurrentLabel(userCreatorEngineTypeLabel);
+        labelContainer.setCurrentLabel(combinedLabel);
         if (!requestResourceService.canRequest(labelContainer, resource)) {
           return new NotEnoughResource(
-              String.format(
-                  "Labels:%s not enough resource", userCreatorEngineTypeLabel.getStringValue()));
+              String.format("Labels:%s not enough resource", combinedLabel.getStringValue()));
         }
       } catch (RMWarnException exception) {
         return new NotEnoughResource(exception.getMessage());
@@ -294,9 +291,7 @@ public ResultResource requestResource(List<Label<?>> labels, NodeResource resour
           labelResource.setLockedResource(
               labelResource.getLockedResource().add(resource.getLockedResource()));
           labelResourceService.setLabelResource(
-              label,
-              labelResource,
-              labelContainer.getCombinedUserCreatorEngineTypeLabel().getStringValue());
+              label, labelResource, labelContainer.getCombinedResourceLabel().getStringValue());
           logger.info(
               String.format(
                   "ResourceChanged:%s --> %s", label.getStringValue(), labelResource.toString()));
@@ -316,6 +311,7 @@ public ResultResource requestResource(List<Label<?>> labels, NodeResource resour
     engineNode.setServiceInstance(
         ServiceInstance.apply(labelContainer.getEngineServiceName(), tickedId));
     engineNode.setNodeResource(resource);
+    engineNode.setTicketId(tickedId);
 
     nodeManagerPersistence.addEngineNode(engineNode);
 
@@ -329,9 +325,7 @@ public ResultResource requestResource(List<Label<?>> labels, NodeResource resour
 
     // add ec resource
     labelResourceService.setEngineConnLabelResource(
-        engineInstanceLabel,
-        resource,
-        labelContainer.getCombinedUserCreatorEngineTypeLabel().getStringValue());
+        engineInstanceLabel, resource, labelContainer.getCombinedResourceLabel().getStringValue());
     // record engine locked resource
     labelContainer.getLabels().add(engineInstanceLabel);
     resourceLogService.recordUserResourceAction(
@@ -440,7 +434,7 @@ public void resourceUsed(List<Label<?>> labels, NodeResource usedResource) {
       labelResourceService.setLabelResource(
           engineInstanceLabel,
           lockedResource,
-          labelContainer.getCombinedUserCreatorEngineTypeLabel().getStringValue());
+          labelContainer.getCombinedResourceLabel().getStringValue());
       resourceLogService.success(
           ChangeType.ENGINE_INIT, lockedResource.getLockedResource(), engineInstanceLabel, null);
     } catch (Exception exception) {
@@ -480,7 +474,7 @@ public void resourceUsed(List<Label<?>> labels, NodeResource usedResource) {
                 labelResourceService.setLabelResource(
                     label,
                     labelResource,
-                    labelContainer.getCombinedUserCreatorEngineTypeLabel().getStringValue());
+                    labelContainer.getCombinedResourceLabel().getStringValue());
                 labelResourceSet.add(
                     new LabelResourceMapping(label, addedResource, ResourceOperationType.USED));
                 resourceCheck(label, labelResource);
@@ -492,7 +486,7 @@ public void resourceUsed(List<Label<?>> labels, NodeResource usedResource) {
 
         if (label
             .getClass()
-            .isAssignableFrom(labelContainer.getCombinedUserCreatorEngineTypeLabel().getClass())) {
+            .isAssignableFrom(labelContainer.getCombinedResourceLabel().getClass())) {
           resourceLogService.recordUserResourceAction(
               labelContainer,
               persistenceResource.getTicketId(),
@@ -703,9 +697,7 @@ public void resourceReleased(EngineNode ecNode) {
                         labelResourceService.setLabelResource(
                             label,
                             labelResource,
-                            labelContainer
-                                .getCombinedUserCreatorEngineTypeLabel()
-                                .getStringValue());
+                            labelContainer.getCombinedResourceLabel().getStringValue());
                         resourceCheck(label, labelResource);
                       }
                     },
@@ -724,8 +716,7 @@ public void resourceReleased(EngineNode ecNode) {
 
                 if (label
                     .getClass()
-                    .isAssignableFrom(
-                        labelContainer.getCombinedUserCreatorEngineTypeLabel().getClass())) {
+                    .isAssignableFrom(labelContainer.getCombinedResourceLabel().getClass())) {
                   resourceLogService.recordUserResourceAction(
                       labelContainer,
                       persistenceResource.getTicketId(),
diff --git a/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/rm/service/impl/ResourceLogService.java b/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/rm/service/impl/ResourceLogService.java
index dd563f5a92..c5890132c9 100644
--- a/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/rm/service/impl/ResourceLogService.java
+++ b/linkis-computation-governance/linkis-manager/linkis-application-manager/src/main/java/org/apache/linkis/manager/rm/service/impl/ResourceLogService.java
@@ -150,11 +150,10 @@ public void recordUserResourceAction(
     if (RMUtils.RM_RESOURCE_ACTION_RECORD.getValue()) {
       LinkisUtils.tryAndWarn(
           () -> {
-            CombinedLabel userCreatorEngineType =
-                labelContainer.getCombinedUserCreatorEngineTypeLabel();
+            CombinedLabel combinedLabel = labelContainer.getCombinedResourceLabel();
             EngineInstanceLabel engineInstanceLabel = labelContainer.getEngineInstanceLabel();
             EMInstanceLabel eMInstanceLabel = labelContainer.getEMInstanceLabel();
-            if (userCreatorEngineType == null) {
+            if (combinedLabel == null) {
               return;
             }
             ECResourceInfoRecord ecResourceInfoRecord =
@@ -171,11 +170,7 @@ public void recordUserResourceAction(
                       : "";
               ecResourceInfoRecord =
                   new ECResourceInfoRecord(
-                      userCreatorEngineType.getStringValue(),
-                      user,
-                      ticketId,
-                      resource,
-                      logDirSuffix);
+                      combinedLabel.getStringValue(), user, ticketId, resource, logDirSuffix);
               ecResourceRecordMapper.insertECResourceInfoRecord(ecResourceInfoRecord);
             }
             if (engineInstanceLabel != null) {
diff --git a/linkis-computation-governance/linkis-manager/linkis-label-common/src/main/java/org/apache/linkis/manager/label/conf/LabelCommonConfig.java b/linkis-computation-governance/linkis-manager/linkis-label-common/src/main/java/org/apache/linkis/manager/label/conf/LabelCommonConfig.java
index 19747f510d..f0f96ba1f1 100644
--- a/linkis-computation-governance/linkis-manager/linkis-label-common/src/main/java/org/apache/linkis/manager/label/conf/LabelCommonConfig.java
+++ b/linkis-computation-governance/linkis-manager/linkis-label-common/src/main/java/org/apache/linkis/manager/label/conf/LabelCommonConfig.java
@@ -69,6 +69,9 @@ public class LabelCommonConfig {
   public static final CommonVars<String> DATAX_ENGINE_VERSION =
       CommonVars.apply("wds.linkis.datax.engine.version", "3.0.0");
 
+  public static final CommonVars<String> NEBULA_ENGINE_VERSION =
+      CommonVars.apply("wds.linkis.nebula.engine.version", "3.0.0");
+
   public static final CommonVars<String> PRESTO_ENGINE_VERSION =
       CommonVars.apply("wds.linkis.presto.engine.version", "0.234");
 
diff --git a/linkis-computation-governance/linkis-manager/linkis-label-common/src/main/java/org/apache/linkis/manager/label/entity/engine/EngineType.scala b/linkis-computation-governance/linkis-manager/linkis-label-common/src/main/java/org/apache/linkis/manager/label/entity/engine/EngineType.scala
index e5a5f88132..27540fac68 100644
--- a/linkis-computation-governance/linkis-manager/linkis-label-common/src/main/java/org/apache/linkis/manager/label/entity/engine/EngineType.scala
+++ b/linkis-computation-governance/linkis-manager/linkis-label-common/src/main/java/org/apache/linkis/manager/label/entity/engine/EngineType.scala
@@ -47,6 +47,8 @@ object EngineType extends Enumeration with Logging {
 
   val PRESTO = Value("presto")
 
+  val NEBULA = Value("nebula")
+
   val FLINK = Value("flink")
 
   val APPCONN = Value("appconn")
@@ -92,6 +94,7 @@ object EngineType extends Enumeration with Logging {
     case _ if IO_ENGINE_HDFS.toString.equalsIgnoreCase(str) => IO_ENGINE_HDFS
     case _ if PIPELINE.toString.equalsIgnoreCase(str) => PIPELINE
     case _ if PRESTO.toString.equalsIgnoreCase(str) => PRESTO
+    case _ if NEBULA.toString.equalsIgnoreCase(str) => NEBULA
     case _ if FLINK.toString.equalsIgnoreCase(str) => FLINK
     case _ if APPCONN.toString.equals(str) => APPCONN
     case _ if SQOOP.toString.equalsIgnoreCase(str) => SQOOP
diff --git a/linkis-computation-governance/linkis-manager/linkis-label-common/src/main/java/org/apache/linkis/manager/label/entity/engine/RunType.scala b/linkis-computation-governance/linkis-manager/linkis-label-common/src/main/java/org/apache/linkis/manager/label/entity/engine/RunType.scala
index 21a067ed45..abb3e010f8 100644
--- a/linkis-computation-governance/linkis-manager/linkis-label-common/src/main/java/org/apache/linkis/manager/label/entity/engine/RunType.scala
+++ b/linkis-computation-governance/linkis-manager/linkis-label-common/src/main/java/org/apache/linkis/manager/label/entity/engine/RunType.scala
@@ -35,6 +35,7 @@ object RunType extends Enumeration {
   val PIPELINE = Value("pipeline")
   val JDBC = Value("jdbc")
   val PRESTO_SQL = Value("psql")
+  val NEBULA_SQL = Value("ngql")
   val JAR = Value("jar")
   val APPCONN = Value("appconn")
   val FUNCTION_MDQ_TYPE = Value("function.mdq")
diff --git a/linkis-computation-governance/linkis-manager/linkis-label-common/src/main/java/org/apache/linkis/manager/label/utils/EngineTypeLabelCreator.java b/linkis-computation-governance/linkis-manager/linkis-label-common/src/main/java/org/apache/linkis/manager/label/utils/EngineTypeLabelCreator.java
index ee90f40232..40fd7c78b1 100644
--- a/linkis-computation-governance/linkis-manager/linkis-label-common/src/main/java/org/apache/linkis/manager/label/utils/EngineTypeLabelCreator.java
+++ b/linkis-computation-governance/linkis-manager/linkis-label-common/src/main/java/org/apache/linkis/manager/label/utils/EngineTypeLabelCreator.java
@@ -72,6 +72,8 @@ private static void init() {
 
           defaultVersion.put(
               EngineType.HBASE().toString(), LabelCommonConfig.HBASE_ENGINE_VERSION.getValue());
+          defaultVersion.put(
+              EngineType.NEBULA().toString(), LabelCommonConfig.NEBULA_ENGINE_VERSION.getValue());
           defaultVersion.put(
               EngineType.SQOOP().toString(), LabelCommonConfig.SQOOP_ENGINE_VERSION.getValue());
           defaultVersion.put(
diff --git a/linkis-computation-governance/linkis-manager/linkis-label-common/src/main/scala/org/apache/linkis/manager/label/utils/LabelUtil.scala b/linkis-computation-governance/linkis-manager/linkis-label-common/src/main/scala/org/apache/linkis/manager/label/utils/LabelUtil.scala
index 3965a5ea11..986f130686 100644
--- a/linkis-computation-governance/linkis-manager/linkis-label-common/src/main/scala/org/apache/linkis/manager/label/utils/LabelUtil.scala
+++ b/linkis-computation-governance/linkis-manager/linkis-label-common/src/main/scala/org/apache/linkis/manager/label/utils/LabelUtil.scala
@@ -17,6 +17,7 @@
 
 package org.apache.linkis.manager.label.utils
 
+import org.apache.linkis.manager.label.constant.LabelValueConstant
 import org.apache.linkis.manager.label.entity.Label
 import org.apache.linkis.manager.label.entity.engine.{
   CodeLanguageLabel,
@@ -135,4 +136,13 @@ object LabelUtil {
     null.asInstanceOf[A]
   }
 
+  def isYarnClusterMode(labels: util.List[Label[_]]): Boolean = {
+    val label = LabelUtil.getEngingeConnRuntimeModeLabel(labels)
+    val isYarnClusterMode: Boolean = {
+      if (null != label && label.getModeValue.equals(LabelValueConstant.YARN_CLUSTER_VALUE)) true
+      else false
+    }
+    isYarnClusterMode
+  }
+
 }
diff --git a/linkis-computation-governance/linkis-manager/linkis-manager-common/src/main/java/org/apache/linkis/manager/common/entity/resource/CPUResource.java b/linkis-computation-governance/linkis-manager/linkis-manager-common/src/main/java/org/apache/linkis/manager/common/entity/resource/CPUResource.java
index d985dcbc46..aecd3ee3c4 100644
--- a/linkis-computation-governance/linkis-manager/linkis-manager-common/src/main/java/org/apache/linkis/manager/common/entity/resource/CPUResource.java
+++ b/linkis-computation-governance/linkis-manager/linkis-manager-common/src/main/java/org/apache/linkis/manager/common/entity/resource/CPUResource.java
@@ -120,6 +120,12 @@ public boolean less(Resource r) {
     return !notLess(r);
   }
 
+  @Override
+  public int compare(Resource r) {
+    CPUResource cpuResource = toCPUResource(r);
+    return Integer.compare(this.getCores(), cpuResource.getCores());
+  }
+
   @Override
   public String toJson() {
     return String.format(" \"cpu\":%s ", cores);
diff --git a/linkis-computation-governance/linkis-manager/linkis-manager-common/src/main/java/org/apache/linkis/manager/common/entity/resource/DriverAndKubernetesResource.java b/linkis-computation-governance/linkis-manager/linkis-manager-common/src/main/java/org/apache/linkis/manager/common/entity/resource/DriverAndKubernetesResource.java
index 6d047b33a5..3426e91e5c 100644
--- a/linkis-computation-governance/linkis-manager/linkis-manager-common/src/main/java/org/apache/linkis/manager/common/entity/resource/DriverAndKubernetesResource.java
+++ b/linkis-computation-governance/linkis-manager/linkis-manager-common/src/main/java/org/apache/linkis/manager/common/entity/resource/DriverAndKubernetesResource.java
@@ -200,6 +200,42 @@ public boolean less(Resource r) {
     return !notLess(r);
   }
 
+  @Override
+  public int compare(Resource resource) {
+    DriverAndKubernetesResource r = new DriverAndKubernetesResource(resource);
+    if (isModuleOperate(r)) {
+      return loadInstanceResource.compare(r.loadInstanceResource);
+    } else {
+      if (loadInstanceResource.getMemory() > r.loadInstanceResource.getMemory()) {
+        return 1;
+      } else if (loadInstanceResource.getMemory() < r.loadInstanceResource.getMemory()) {
+        return -1;
+      } else {
+        // If memory is equal, compare cores
+        if (loadInstanceResource.getCores() > r.loadInstanceResource.getCores()) {
+          return 1;
+        } else if (loadInstanceResource.getCores() < r.loadInstanceResource.getCores()) {
+          return -1;
+        } else {
+          // If cores are equal, compare instances
+          if (loadInstanceResource.getInstances() > r.loadInstanceResource.getInstances()) {
+            return 1;
+          } else if (loadInstanceResource.getInstances() < r.loadInstanceResource.getInstances()) {
+            return -1;
+          } else {
+            if (kubernetesResource.getMemory() > r.kubernetesResource.getMemory()) {
+              return 1;
+            } else if (kubernetesResource.getMemory() < r.kubernetesResource.getMemory()) {
+              return -1;
+            } else {
+              return Long.compare(kubernetesResource.getCores(), r.kubernetesResource.getCores());
+            }
+          }
+        }
+      }
+    }
+  }
+
   public String toJson() {
     String load = "null";
     String kubernetes = "null";
diff --git a/linkis-computation-governance/linkis-manager/linkis-manager-common/src/main/java/org/apache/linkis/manager/common/entity/resource/DriverAndYarnResource.java b/linkis-computation-governance/linkis-manager/linkis-manager-common/src/main/java/org/apache/linkis/manager/common/entity/resource/DriverAndYarnResource.java
index 4aff24bac0..c9cfcf378e 100644
--- a/linkis-computation-governance/linkis-manager/linkis-manager-common/src/main/java/org/apache/linkis/manager/common/entity/resource/DriverAndYarnResource.java
+++ b/linkis-computation-governance/linkis-manager/linkis-manager-common/src/main/java/org/apache/linkis/manager/common/entity/resource/DriverAndYarnResource.java
@@ -202,6 +202,43 @@ public boolean less(Resource r) {
     return !notLess(r);
   }
 
+  @Override
+  public int compare(Resource resource) {
+    DriverAndYarnResource r = new DriverAndYarnResource(resource);
+
+    if (isModuleOperate(r)) {
+      return loadInstanceResource.compare(r.loadInstanceResource);
+    } else {
+      if (loadInstanceResource.getMemory() > r.loadInstanceResource.getMemory()) {
+        return 1;
+      } else if (loadInstanceResource.getMemory() < r.loadInstanceResource.getMemory()) {
+        return -1;
+      } else {
+        // If memory is equal, compare cores
+        if (loadInstanceResource.getCores() > r.loadInstanceResource.getCores()) {
+          return 1;
+        } else if (loadInstanceResource.getCores() < r.loadInstanceResource.getCores()) {
+          return -1;
+        } else {
+          // If cores are equal, compare instances
+          if (loadInstanceResource.getInstances() > r.loadInstanceResource.getInstances()) {
+            return 1;
+          } else if (loadInstanceResource.getInstances() < r.loadInstanceResource.getInstances()) {
+            return -1;
+          } else {
+            if (yarnResource.getQueueMemory() > r.yarnResource.getQueueMemory()) {
+              return 1;
+            } else if (yarnResource.getQueueMemory() < r.yarnResource.getQueueMemory()) {
+              return -1;
+            } else {
+              return Integer.compare(yarnResource.getQueueCores(), r.yarnResource.getQueueCores());
+            }
+          }
+        }
+      }
+    }
+  }
+
   public String toJson() {
     String load = "null";
     String yarn = "null";
diff --git a/linkis-computation-governance/linkis-manager/linkis-manager-common/src/main/java/org/apache/linkis/manager/common/entity/resource/KubernetesResource.java b/linkis-computation-governance/linkis-manager/linkis-manager-common/src/main/java/org/apache/linkis/manager/common/entity/resource/KubernetesResource.java
index d716569c55..c247620685 100644
--- a/linkis-computation-governance/linkis-manager/linkis-manager-common/src/main/java/org/apache/linkis/manager/common/entity/resource/KubernetesResource.java
+++ b/linkis-computation-governance/linkis-manager/linkis-manager-common/src/main/java/org/apache/linkis/manager/common/entity/resource/KubernetesResource.java
@@ -127,6 +127,19 @@ public boolean less(Resource r) {
     return !notLess(r);
   }
 
+  @Override
+  public int compare(Resource r) {
+    KubernetesResource temp = new KubernetesResource(r);
+    if (this.getMemory() > temp.getMemory()) {
+      return 1;
+    } else if (this.getMemory() < temp.getMemory()) {
+      return -1;
+    } else {
+      // If memory is equal, compare cores
+      return Long.compare(this.getCores(), temp.getCores());
+    }
+  }
+
   @Override
   public String toJson() {
     return String.format(
diff --git a/linkis-computation-governance/linkis-manager/linkis-manager-common/src/main/java/org/apache/linkis/manager/common/entity/resource/LoadInstanceResource.java b/linkis-computation-governance/linkis-manager/linkis-manager-common/src/main/java/org/apache/linkis/manager/common/entity/resource/LoadInstanceResource.java
index 57ccc4f313..2eb37cc551 100644
--- a/linkis-computation-governance/linkis-manager/linkis-manager-common/src/main/java/org/apache/linkis/manager/common/entity/resource/LoadInstanceResource.java
+++ b/linkis-computation-governance/linkis-manager/linkis-manager-common/src/main/java/org/apache/linkis/manager/common/entity/resource/LoadInstanceResource.java
@@ -138,6 +138,27 @@ public boolean less(Resource r) {
     return !notLess(r);
   }
 
+  @Override
+  public int compare(Resource r) {
+    LoadInstanceResource temp = new LoadInstanceResource(r);
+
+    if (this.getMemory() > temp.getMemory()) {
+      return 1;
+    } else if (this.getMemory() < temp.getMemory()) {
+      return -1;
+    } else {
+      // If memory is equal, compare cores
+      if (this.getCores() > temp.getCores()) {
+        return 1;
+      } else if (this.getCores() < temp.getCores()) {
+        return -1;
+      } else {
+        // If cores are equal, compare instances
+        return Integer.compare(this.getInstances(), temp.getInstances());
+      }
+    }
+  }
+
   @Override
   public String toJson() {
     return String.format(
@@ -149,7 +170,7 @@ public String toJson() {
   public String toString() {
     return String.format(
         "Number of instances(实例数)：%d，(RAM)内存：%s ,cpu: %s",
-        this.getInstances(), this.getCores(), this.getMemory());
+        this.getInstances(), this.getMemory(), this.getCores());
   }
 
   public long getMemory() {
diff --git a/linkis-computation-governance/linkis-manager/linkis-manager-common/src/main/java/org/apache/linkis/manager/common/entity/resource/LoadResource.java b/linkis-computation-governance/linkis-manager/linkis-manager-common/src/main/java/org/apache/linkis/manager/common/entity/resource/LoadResource.java
index 79a6374844..313ce7b92b 100644
--- a/linkis-computation-governance/linkis-manager/linkis-manager-common/src/main/java/org/apache/linkis/manager/common/entity/resource/LoadResource.java
+++ b/linkis-computation-governance/linkis-manager/linkis-manager-common/src/main/java/org/apache/linkis/manager/common/entity/resource/LoadResource.java
@@ -108,6 +108,20 @@ public boolean less(Resource r) {
     return !notLess(r);
   }
 
+  @Override
+  public int compare(Resource r) {
+    LoadResource temp = new LoadResource(r);
+
+    if (this.getMemory() > temp.getMemory()) {
+      return 1;
+    } else if (this.getMemory() < temp.getMemory()) {
+      return -1;
+    } else {
+      // If memory is equal, compare cores
+      return Integer.compare(this.getCores(), temp.getCores());
+    }
+  }
+
   @Override
   public String toJson() {
     return String.format(
diff --git a/linkis-computation-governance/linkis-manager/linkis-manager-common/src/main/java/org/apache/linkis/manager/common/entity/resource/MemoryResource.java b/linkis-computation-governance/linkis-manager/linkis-manager-common/src/main/java/org/apache/linkis/manager/common/entity/resource/MemoryResource.java
index 23dc96fdb0..da69c930c7 100644
--- a/linkis-computation-governance/linkis-manager/linkis-manager-common/src/main/java/org/apache/linkis/manager/common/entity/resource/MemoryResource.java
+++ b/linkis-computation-governance/linkis-manager/linkis-manager-common/src/main/java/org/apache/linkis/manager/common/entity/resource/MemoryResource.java
@@ -93,6 +93,12 @@ public boolean less(Resource r) {
     return !notLess(r);
   }
 
+  @Override
+  public int compare(Resource resource) {
+    MemoryResource r = toMemoryResource(resource);
+    return Long.compare(this.getMemory(), r.getMemory());
+  }
+
   @Override
   public boolean caseMore(Resource r) {
     return moreThan(r);
diff --git a/linkis-computation-governance/linkis-manager/linkis-manager-common/src/main/java/org/apache/linkis/manager/common/entity/resource/Resource.java b/linkis-computation-governance/linkis-manager/linkis-manager-common/src/main/java/org/apache/linkis/manager/common/entity/resource/Resource.java
index 361a6f28e0..ca45f374c5 100644
--- a/linkis-computation-governance/linkis-manager/linkis-manager-common/src/main/java/org/apache/linkis/manager/common/entity/resource/Resource.java
+++ b/linkis-computation-governance/linkis-manager/linkis-manager-common/src/main/java/org/apache/linkis/manager/common/entity/resource/Resource.java
@@ -122,6 +122,8 @@ public static Resource getZeroResource(Resource resource) {
 
   public abstract boolean less(Resource r);
 
+  public abstract int compare(Resource r);
+
   public Resource add(Resource r, float rate) {
     return this.add(r.multiplied(rate));
   }
diff --git a/linkis-computation-governance/linkis-manager/linkis-manager-common/src/main/java/org/apache/linkis/manager/common/entity/resource/SpecialResource.java b/linkis-computation-governance/linkis-manager/linkis-manager-common/src/main/java/org/apache/linkis/manager/common/entity/resource/SpecialResource.java
index b670eb174e..60518a1552 100644
--- a/linkis-computation-governance/linkis-manager/linkis-manager-common/src/main/java/org/apache/linkis/manager/common/entity/resource/SpecialResource.java
+++ b/linkis-computation-governance/linkis-manager/linkis-manager-common/src/main/java/org/apache/linkis/manager/common/entity/resource/SpecialResource.java
@@ -348,6 +348,17 @@ public boolean less(Resource r) {
     return !notLess(r);
   }
 
+  @Override
+  public int compare(Resource r) {
+    if (this.moreThan(r)) {
+      return 1;
+    } else if (this.less(r)) {
+      return -1;
+    } else {
+      return 0;
+    }
+  }
+
   @Override
   public String toJson() {
     return String.format("Special:%s", resources);
diff --git a/linkis-computation-governance/linkis-manager/linkis-manager-common/src/main/java/org/apache/linkis/manager/common/entity/resource/YarnResource.java b/linkis-computation-governance/linkis-manager/linkis-manager-common/src/main/java/org/apache/linkis/manager/common/entity/resource/YarnResource.java
index 25f4d82984..48a63cd0f5 100644
--- a/linkis-computation-governance/linkis-manager/linkis-manager-common/src/main/java/org/apache/linkis/manager/common/entity/resource/YarnResource.java
+++ b/linkis-computation-governance/linkis-manager/linkis-manager-common/src/main/java/org/apache/linkis/manager/common/entity/resource/YarnResource.java
@@ -160,6 +160,18 @@ public boolean less(Resource r) {
     return !notLess(r);
   }
 
+  @Override
+  public int compare(Resource resource) {
+    YarnResource r = toYarnResource(resource);
+    if (this.getQueueMemory() > r.getQueueMemory()) {
+      return 1;
+    } else if (this.getQueueMemory() < r.getQueueMemory()) {
+      return -1;
+    } else {
+      return Integer.compare(this.getQueueCores(), r.getQueueCores());
+    }
+  }
+
   @Override
   public String toJson() {
     return String.format(
diff --git a/linkis-engineconn-plugins/nebula/pom.xml b/linkis-engineconn-plugins/nebula/pom.xml
new file mode 100644
index 0000000000..bfe9714569
--- /dev/null
+++ b/linkis-engineconn-plugins/nebula/pom.xml
@@ -0,0 +1,110 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ 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.
+  -->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+  <modelVersion>4.0.0</modelVersion>
+  <parent>
+    <groupId>org.apache.linkis</groupId>
+    <artifactId>linkis</artifactId>
+    <version>${revision}</version>
+    <relativePath>../../pom.xml</relativePath>
+  </parent>
+
+  <artifactId>linkis-engineplugin-nebula</artifactId>
+
+  <dependencies>
+    <dependency>
+      <groupId>org.apache.linkis</groupId>
+      <artifactId>linkis-engineconn-plugin-core</artifactId>
+      <version>${project.version}</version>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.linkis</groupId>
+      <artifactId>linkis-computation-engineconn</artifactId>
+      <version>${project.version}</version>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.linkis</groupId>
+      <artifactId>linkis-storage</artifactId>
+      <version>${project.version}</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.linkis</groupId>
+      <artifactId>linkis-rpc</artifactId>
+      <version>${project.version}</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.linkis</groupId>
+      <artifactId>linkis-common</artifactId>
+      <version>${project.version}</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <!-- nebula -->
+    <dependency>
+      <groupId>com.vesoft</groupId>
+      <artifactId>client</artifactId>
+      <version>${nebula.version}</version>
+    </dependency>
+  </dependencies>
+
+  <build>
+
+    <plugins>
+      <plugin>
+        <groupId>net.alchim31.maven</groupId>
+        <artifactId>scala-maven-plugin</artifactId>
+      </plugin>
+
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-assembly-plugin</artifactId>
+        <inherited>false</inherited>
+        <configuration>
+          <skipAssembly>false</skipAssembly>
+          <finalName>out</finalName>
+          <appendAssemblyId>false</appendAssemblyId>
+          <attach>false</attach>
+          <descriptors>
+            <descriptor>src/main/assembly/distribution.xml</descriptor>
+          </descriptors>
+        </configuration>
+        <executions>
+          <execution>
+            <id>make-assembly</id>
+            <goals>
+              <goal>single</goal>
+            </goals>
+            <phase>package</phase>
+            <configuration>
+              <descriptors>
+                <descriptor>src/main/assembly/distribution.xml</descriptor>
+              </descriptors>
+            </configuration>
+          </execution>
+        </executions>
+      </plugin>
+    </plugins>
+  </build>
+
+</project>
diff --git a/linkis-engineconn-plugins/nebula/src/main/assembly/distribution.xml b/linkis-engineconn-plugins/nebula/src/main/assembly/distribution.xml
new file mode 100644
index 0000000000..eaa9c296f1
--- /dev/null
+++ b/linkis-engineconn-plugins/nebula/src/main/assembly/distribution.xml
@@ -0,0 +1,71 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ 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.
+  -->
+
+<assembly xmlns="http://maven.apache.org/ASSEMBLY/2.1.1" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+          xsi:schemaLocation="http://maven.apache.org/ASSEMBLY/2.1.1 https://maven.apache.org/xsd/assembly-2.1.1.xsd">
+    <id>linkis-engineplugin-nebula</id>
+    <formats>
+        <format>dir</format>
+        <format>zip</format>
+    </formats>
+    <includeBaseDirectory>true</includeBaseDirectory>
+    <baseDirectory>nebula</baseDirectory>
+
+    <dependencySets>
+        <dependencySet>
+            <!-- Enable access to all projects in the current multimodule build! <useAllReactorProjects>true</useAllReactorProjects> -->
+            <!-- Now, select which projects to include in this module-set. -->
+            <outputDirectory>/dist/${nebula.version}/lib</outputDirectory>
+            <useProjectArtifact>true</useProjectArtifact>
+            <useTransitiveDependencies>true</useTransitiveDependencies>
+            <unpack>false</unpack>
+            <useStrictFiltering>false</useStrictFiltering>
+            <useTransitiveFiltering>true</useTransitiveFiltering>
+
+        </dependencySet>
+    </dependencySets>
+
+    <fileSets>
+
+        <fileSet>
+            <directory>${basedir}/src/main/resources</directory>
+            <includes>
+                <include>linkis-engineconn.properties</include>
+                <include>log4j2.xml</include>
+            </includes>
+            <fileMode>0777</fileMode>
+            <outputDirectory>dist/${nebula.version}/conf</outputDirectory>
+            <lineEnding>unix</lineEnding>
+        </fileSet>
+
+        <fileSet>
+            <directory>${basedir}/target</directory>
+            <includes>
+                <include>*.jar</include>
+            </includes>
+            <excludes>
+                <exclude>*doc.jar</exclude>
+            </excludes>
+            <fileMode>0777</fileMode>
+            <outputDirectory>plugin/${nebula.version}</outputDirectory>
+        </fileSet>
+
+    </fileSets>
+
+</assembly>
+
diff --git a/linkis-engineconn-plugins/nebula/src/main/java/org/apache/linkis/engineplugin/nebula/NebulaEngineConnPlugin.java b/linkis-engineconn-plugins/nebula/src/main/java/org/apache/linkis/engineplugin/nebula/NebulaEngineConnPlugin.java
new file mode 100644
index 0000000000..a22d2c8a84
--- /dev/null
+++ b/linkis-engineconn-plugins/nebula/src/main/java/org/apache/linkis/engineplugin/nebula/NebulaEngineConnPlugin.java
@@ -0,0 +1,72 @@
+/*
+ * 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.linkis.engineplugin.nebula;
+
+import org.apache.linkis.engineplugin.nebula.builder.NebulaProcessEngineConnLaunchBuilder;
+import org.apache.linkis.engineplugin.nebula.factory.NebulaEngineConnFactory;
+import org.apache.linkis.manager.engineplugin.common.EngineConnPlugin;
+import org.apache.linkis.manager.engineplugin.common.creation.EngineConnFactory;
+import org.apache.linkis.manager.engineplugin.common.launch.EngineConnLaunchBuilder;
+import org.apache.linkis.manager.engineplugin.common.resource.EngineResourceFactory;
+import org.apache.linkis.manager.engineplugin.common.resource.GenericEngineResourceFactory;
+import org.apache.linkis.manager.label.entity.Label;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+public class NebulaEngineConnPlugin implements EngineConnPlugin {
+  private Object resourceLocker = new Object();
+  private Object engineFactoryLocker = new Object();
+  private volatile EngineResourceFactory engineResourceFactory;
+  private volatile EngineConnFactory engineFactory;
+  private List<Label<?>> defaultLabels = new ArrayList<>();
+
+  @Override
+  public void init(Map<String, Object> params) {}
+
+  @Override
+  public EngineResourceFactory getEngineResourceFactory() {
+    if (null == engineResourceFactory) {
+      synchronized (resourceLocker) {
+        engineResourceFactory = new GenericEngineResourceFactory();
+      }
+    }
+    return engineResourceFactory;
+  }
+
+  @Override
+  public EngineConnLaunchBuilder getEngineConnLaunchBuilder() {
+    return new NebulaProcessEngineConnLaunchBuilder();
+  }
+
+  @Override
+  public EngineConnFactory getEngineConnFactory() {
+    if (null == engineFactory) {
+      synchronized (engineFactoryLocker) {
+        engineFactory = new NebulaEngineConnFactory();
+      }
+    }
+    return engineFactory;
+  }
+
+  @Override
+  public List<Label<?>> getDefaultLabels() {
+    return defaultLabels;
+  }
+}
diff --git a/linkis-engineconn-plugins/nebula/src/main/java/org/apache/linkis/engineplugin/nebula/builder/NebulaProcessEngineConnLaunchBuilder.java b/linkis-engineconn-plugins/nebula/src/main/java/org/apache/linkis/engineplugin/nebula/builder/NebulaProcessEngineConnLaunchBuilder.java
new file mode 100644
index 0000000000..fb95910cf5
--- /dev/null
+++ b/linkis-engineconn-plugins/nebula/src/main/java/org/apache/linkis/engineplugin/nebula/builder/NebulaProcessEngineConnLaunchBuilder.java
@@ -0,0 +1,22 @@
+/*
+ * 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.linkis.engineplugin.nebula.builder;
+
+import org.apache.linkis.manager.engineplugin.common.launch.process.JavaProcessEngineConnLaunchBuilder;
+
+public class NebulaProcessEngineConnLaunchBuilder extends JavaProcessEngineConnLaunchBuilder {}
diff --git a/linkis-engineconn-plugins/nebula/src/main/java/org/apache/linkis/engineplugin/nebula/conf/NebulaConfiguration.java b/linkis-engineconn-plugins/nebula/src/main/java/org/apache/linkis/engineplugin/nebula/conf/NebulaConfiguration.java
new file mode 100644
index 0000000000..dfbb7a8b13
--- /dev/null
+++ b/linkis-engineconn-plugins/nebula/src/main/java/org/apache/linkis/engineplugin/nebula/conf/NebulaConfiguration.java
@@ -0,0 +1,50 @@
+/*
+ * 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.linkis.engineplugin.nebula.conf;
+
+import org.apache.linkis.common.conf.CommonVars;
+
+public class NebulaConfiguration {
+
+  public static final CommonVars<Integer> ENGINE_CONCURRENT_LIMIT =
+      CommonVars.apply("linkis.engineconn.concurrent.limit", 100);
+
+  public static final CommonVars<Integer> ENGINE_DEFAULT_LIMIT =
+      CommonVars.apply("linkis.nebula.default.limit", 5000);
+
+  public static final CommonVars<String> NEBULA_HOST =
+      CommonVars.apply("linkis.nebula.host", "127.0.0.1");
+
+  public static final CommonVars<Integer> NEBULA_PORT =
+      CommonVars.apply("linkis.nebula.port", 9669);
+
+  public static final CommonVars<Integer> NEBULA_MAX_CONN_SIZE =
+      CommonVars.apply("linkis.nebula.max.conn.size", 100);
+
+  public static final CommonVars<String> NEBULA_USER_NAME =
+      CommonVars.apply("linkis.nebula.username", "root");
+
+  public static final CommonVars<String> NEBULA_PASSWORD =
+      CommonVars.apply("linkis.nebula.password", "nebula");
+
+  public static final CommonVars<Boolean> NEBULA_RECONNECT_ENABLED =
+      CommonVars.apply(
+          "linkis.nebula.reconnect.enabled",
+          false,
+          "whether to retry after the connection is disconnected");
+}
diff --git a/linkis-engineconn-plugins/nebula/src/main/java/org/apache/linkis/engineplugin/nebula/conf/NebulaEngineConf.java b/linkis-engineconn-plugins/nebula/src/main/java/org/apache/linkis/engineplugin/nebula/conf/NebulaEngineConf.java
new file mode 100644
index 0000000000..92cc32ca01
--- /dev/null
+++ b/linkis-engineconn-plugins/nebula/src/main/java/org/apache/linkis/engineplugin/nebula/conf/NebulaEngineConf.java
@@ -0,0 +1,53 @@
+/*
+ * 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.linkis.engineplugin.nebula.conf;
+
+import org.apache.linkis.common.conf.Configuration;
+import org.apache.linkis.governance.common.protocol.conf.RequestQueryEngineConfigWithGlobalConfig;
+import org.apache.linkis.governance.common.protocol.conf.ResponseQueryConfig;
+import org.apache.linkis.manager.label.entity.engine.EngineTypeLabel;
+import org.apache.linkis.manager.label.entity.engine.UserCreatorLabel;
+import org.apache.linkis.protocol.CacheableProtocol;
+import org.apache.linkis.rpc.RPCMapCache;
+
+import java.util.Map;
+
+import scala.Tuple2;
+
+public class NebulaEngineConf
+    extends RPCMapCache<Tuple2<UserCreatorLabel, EngineTypeLabel>, String, String> {
+
+  public NebulaEngineConf() {
+    super(Configuration.CLOUD_CONSOLE_CONFIGURATION_SPRING_APPLICATION_NAME().getValue());
+  }
+
+  @Override
+  public CacheableProtocol createRequest(Tuple2<UserCreatorLabel, EngineTypeLabel> labelTuple) {
+    return new RequestQueryEngineConfigWithGlobalConfig(labelTuple._1(), labelTuple._2(), null);
+  }
+
+  @Override
+  public Map<String, String> createMap(Object obj) {
+    if (obj instanceof ResponseQueryConfig) {
+      ResponseQueryConfig response = (ResponseQueryConfig) obj;
+      return response.getKeyAndValue();
+    } else {
+      return null;
+    }
+  }
+}
diff --git a/linkis-engineconn-plugins/nebula/src/main/java/org/apache/linkis/engineplugin/nebula/errorcode/NebulaErrorCodeSummary.java b/linkis-engineconn-plugins/nebula/src/main/java/org/apache/linkis/engineplugin/nebula/errorcode/NebulaErrorCodeSummary.java
new file mode 100644
index 0000000000..80aa2e197e
--- /dev/null
+++ b/linkis-engineconn-plugins/nebula/src/main/java/org/apache/linkis/engineplugin/nebula/errorcode/NebulaErrorCodeSummary.java
@@ -0,0 +1,47 @@
+/*
+ * 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.linkis.engineplugin.nebula.errorcode;
+
+import org.apache.linkis.common.errorcode.ErrorCodeUtils;
+import org.apache.linkis.common.errorcode.LinkisErrorCode;
+
+public enum NebulaErrorCodeSummary implements LinkisErrorCode {
+  NEBULA_CLIENT_INITIALIZATION_FAILED(28001, "Nebula client initialization failed(Nebula客户端初始化失败)"),
+  NEBULA_EXECUTOR_ERROR(28002, "Nebula executor error(Nebula执行异常)"),
+  NEBULA_CLIENT_ERROR(28003, "Nebula client error(Nebula客户端异常)");
+
+  private final int errorCode;
+
+  private final String errorDesc;
+
+  NebulaErrorCodeSummary(int errorCode, String errorDesc) {
+    ErrorCodeUtils.validateErrorCode(errorCode, 26000, 29999);
+    this.errorCode = errorCode;
+    this.errorDesc = errorDesc;
+  }
+
+  @Override
+  public int getErrorCode() {
+    return errorCode;
+  }
+
+  @Override
+  public String getErrorDesc() {
+    return errorDesc;
+  }
+}
diff --git a/linkis-engineconn-plugins/nebula/src/main/java/org/apache/linkis/engineplugin/nebula/exception/NebulaClientException.java b/linkis-engineconn-plugins/nebula/src/main/java/org/apache/linkis/engineplugin/nebula/exception/NebulaClientException.java
new file mode 100644
index 0000000000..59b3620b03
--- /dev/null
+++ b/linkis-engineconn-plugins/nebula/src/main/java/org/apache/linkis/engineplugin/nebula/exception/NebulaClientException.java
@@ -0,0 +1,27 @@
+/*
+ * 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.linkis.engineplugin.nebula.exception;
+
+import org.apache.linkis.common.exception.ErrorException;
+
+public class NebulaClientException extends ErrorException {
+
+  public NebulaClientException(int errorCode, String message) {
+    super(errorCode, message);
+  }
+}
diff --git a/linkis-engineconn-plugins/nebula/src/main/java/org/apache/linkis/engineplugin/nebula/exception/NebulaExecuteError.java b/linkis-engineconn-plugins/nebula/src/main/java/org/apache/linkis/engineplugin/nebula/exception/NebulaExecuteError.java
new file mode 100644
index 0000000000..f2c164d5a2
--- /dev/null
+++ b/linkis-engineconn-plugins/nebula/src/main/java/org/apache/linkis/engineplugin/nebula/exception/NebulaExecuteError.java
@@ -0,0 +1,27 @@
+/*
+ * 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.linkis.engineplugin.nebula.exception;
+
+import org.apache.linkis.common.exception.ErrorException;
+
+public class NebulaExecuteError extends ErrorException {
+
+  public NebulaExecuteError(int errorCode, String message) {
+    super(errorCode, message);
+  }
+}
diff --git a/linkis-engineconn-plugins/nebula/src/main/java/org/apache/linkis/engineplugin/nebula/exception/NebulaStateInvalidException.java b/linkis-engineconn-plugins/nebula/src/main/java/org/apache/linkis/engineplugin/nebula/exception/NebulaStateInvalidException.java
new file mode 100644
index 0000000000..202d478b76
--- /dev/null
+++ b/linkis-engineconn-plugins/nebula/src/main/java/org/apache/linkis/engineplugin/nebula/exception/NebulaStateInvalidException.java
@@ -0,0 +1,27 @@
+/*
+ * 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.linkis.engineplugin.nebula.exception;
+
+import org.apache.linkis.common.exception.ErrorException;
+
+public class NebulaStateInvalidException extends ErrorException {
+
+  public NebulaStateInvalidException(int errorCode, String message) {
+    super(errorCode, message);
+  }
+}
diff --git a/linkis-engineconn-plugins/nebula/src/main/java/org/apache/linkis/engineplugin/nebula/executor/NebulaEngineConnExecutor.java b/linkis-engineconn-plugins/nebula/src/main/java/org/apache/linkis/engineplugin/nebula/executor/NebulaEngineConnExecutor.java
new file mode 100644
index 0000000000..188ea60ec4
--- /dev/null
+++ b/linkis-engineconn-plugins/nebula/src/main/java/org/apache/linkis/engineplugin/nebula/executor/NebulaEngineConnExecutor.java
@@ -0,0 +1,388 @@
+/*
+ * 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.linkis.engineplugin.nebula.executor;
+
+import org.apache.linkis.common.exception.ErrorException;
+import org.apache.linkis.common.io.resultset.ResultSetWriter;
+import org.apache.linkis.common.log.LogUtils;
+import org.apache.linkis.common.utils.OverloadUtils;
+import org.apache.linkis.engineconn.common.conf.EngineConnConf;
+import org.apache.linkis.engineconn.common.conf.EngineConnConstant;
+import org.apache.linkis.engineconn.computation.executor.entity.EngineConnTask;
+import org.apache.linkis.engineconn.computation.executor.execute.ConcurrentComputationExecutor;
+import org.apache.linkis.engineconn.computation.executor.execute.EngineExecutionContext;
+import org.apache.linkis.engineconn.core.EngineConnObject;
+import org.apache.linkis.engineplugin.nebula.conf.NebulaConfiguration;
+import org.apache.linkis.engineplugin.nebula.conf.NebulaEngineConf;
+import org.apache.linkis.engineplugin.nebula.errorcode.NebulaErrorCodeSummary;
+import org.apache.linkis.engineplugin.nebula.exception.NebulaClientException;
+import org.apache.linkis.engineplugin.nebula.exception.NebulaExecuteError;
+import org.apache.linkis.governance.common.paser.SQLCodeParser;
+import org.apache.linkis.manager.common.entity.resource.CommonNodeResource;
+import org.apache.linkis.manager.common.entity.resource.LoadResource;
+import org.apache.linkis.manager.common.entity.resource.NodeResource;
+import org.apache.linkis.manager.engineplugin.common.util.NodeResourceUtils;
+import org.apache.linkis.manager.label.entity.Label;
+import org.apache.linkis.manager.label.entity.engine.EngineTypeLabel;
+import org.apache.linkis.manager.label.entity.engine.UserCreatorLabel;
+import org.apache.linkis.protocol.engine.JobProgressInfo;
+import org.apache.linkis.rpc.Sender;
+import org.apache.linkis.scheduler.executer.ErrorExecuteResponse;
+import org.apache.linkis.scheduler.executer.ExecuteResponse;
+import org.apache.linkis.scheduler.executer.SuccessExecuteResponse;
+import org.apache.linkis.storage.domain.Column;
+import org.apache.linkis.storage.domain.DataType;
+import org.apache.linkis.storage.resultset.ResultSetFactory;
+import org.apache.linkis.storage.resultset.table.TableMetaData;
+import org.apache.linkis.storage.resultset.table.TableRecord;
+
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.lang3.StringUtils;
+
+import org.springframework.util.CollectionUtils;
+
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import scala.Tuple2;
+
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
+import com.vesoft.nebula.ErrorCode;
+import com.vesoft.nebula.client.graph.NebulaPoolConfig;
+import com.vesoft.nebula.client.graph.data.HostAddress;
+import com.vesoft.nebula.client.graph.data.ResultSet;
+import com.vesoft.nebula.client.graph.net.NebulaPool;
+import com.vesoft.nebula.client.graph.net.Session;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class NebulaEngineConnExecutor extends ConcurrentComputationExecutor {
+
+  private static final Logger logger = LoggerFactory.getLogger(NebulaEngineConnExecutor.class);
+  private int id;
+  private List<Label<?>> executorLabels = new ArrayList<>(2);
+  private Map<String, Session> sessionCache = new ConcurrentHashMap<>();
+
+  private Map<String, String> configMap = new HashMap<>();
+
+  private Cache<String, NebulaPool> nebulaPoolCache =
+      CacheBuilder.newBuilder()
+          .expireAfterAccess(
+              Long.valueOf(EngineConnConf.ENGINE_TASK_EXPIRE_TIME().getValue().toString()),
+              TimeUnit.MILLISECONDS)
+          .maximumSize(EngineConnConstant.MAX_TASK_NUM())
+          .build();
+
+  public NebulaEngineConnExecutor(int outputPrintLimit, int id) {
+    super(outputPrintLimit);
+    this.id = id;
+  }
+
+  @Override
+  public void init() {
+    setCodeParser(new SQLCodeParser());
+    super.init();
+  }
+
+  @Override
+  public ExecuteResponse execute(EngineConnTask engineConnTask) {
+    Optional<Label<?>> userCreatorLabelOp =
+        Arrays.stream(engineConnTask.getLables())
+            .filter(label -> label instanceof UserCreatorLabel)
+            .findFirst();
+    Optional<Label<?>> engineTypeLabelOp =
+        Arrays.stream(engineConnTask.getLables())
+            .filter(label -> label instanceof EngineTypeLabel)
+            .findFirst();
+
+    Map<String, String> configMap = null;
+    if (userCreatorLabelOp.isPresent() && engineTypeLabelOp.isPresent()) {
+      UserCreatorLabel userCreatorLabel = (UserCreatorLabel) userCreatorLabelOp.get();
+      EngineTypeLabel engineTypeLabel = (EngineTypeLabel) engineTypeLabelOp.get();
+
+      configMap =
+          new NebulaEngineConf().getCacheMap(new Tuple2<>(userCreatorLabel, engineTypeLabel));
+    }
+
+    nebulaPoolCache.put(
+        engineConnTask.getTaskId(), getNebulaPool(engineConnTask.getProperties(), configMap));
+    return super.execute(engineConnTask);
+  }
+
+  @Override
+  public ExecuteResponse executeLine(EngineExecutionContext engineExecutorContext, String code) {
+    String realCode;
+    if (StringUtils.isBlank(code)) {
+      realCode = "SHOW SPACES";
+    } else {
+      realCode = code.trim();
+    }
+    logger.info("Nebula client begins to run ngql code:\n {}", realCode);
+
+    String taskId = engineExecutorContext.getJobId().get();
+    NebulaPool nebulaPool = nebulaPoolCache.getIfPresent(taskId);
+    Session session = getSession(taskId, nebulaPool);
+
+    initialStatusUpdates(taskId, engineExecutorContext, session);
+    ResultSet resultSet = null;
+
+    try {
+      resultSet = session.execute(code);
+    } catch (Exception e) {
+      logger.error("Nebula executor error.");
+      throw new NebulaExecuteError(
+          NebulaErrorCodeSummary.NEBULA_EXECUTOR_ERROR.getErrorCode(),
+          NebulaErrorCodeSummary.NEBULA_EXECUTOR_ERROR.getErrorDesc());
+    }
+
+    if (resultSet.isSucceeded() && !resultSet.isEmpty()) {
+      queryOutput(taskId, engineExecutorContext, resultSet);
+    }
+    ErrorExecuteResponse errorResponse = null;
+    try {
+      errorResponse = verifyServerError(taskId, engineExecutorContext, resultSet);
+    } catch (ErrorException e) {
+      logger.error("Nebula execute failed (#{}): {}", e.getErrCode(), e.getMessage());
+    }
+    if (errorResponse == null) {
+      return new SuccessExecuteResponse();
+    } else {
+      return errorResponse;
+    }
+  }
+
+  @Override
+  public ExecuteResponse executeCompletely(
+      EngineExecutionContext engineExecutorContext, String code, String completedLine) {
+    return null;
+  }
+
+  @Override
+  public float progress(String taskID) {
+    return 0.0f;
+  }
+
+  @Override
+  public JobProgressInfo[] getProgressInfo(String taskID) {
+    return new JobProgressInfo[0];
+  }
+
+  @Override
+  public void killTask(String taskId) {
+    Session session = sessionCache.remove(taskId);
+    if (null != session) {
+      session.release();
+    }
+    super.killTask(taskId);
+  }
+
+  @Override
+  public List<Label<?>> getExecutorLabels() {
+    return executorLabels;
+  }
+
+  @Override
+  public void setExecutorLabels(List<Label<?>> labels) {
+    if (!CollectionUtils.isEmpty(labels)) {
+      executorLabels.clear();
+      executorLabels.addAll(labels);
+    }
+  }
+
+  @Override
+  public boolean supportCallBackLogs() {
+    return false;
+  }
+
+  @Override
+  public NodeResource requestExpectedResource(NodeResource expectedResource) {
+    return null;
+  }
+
+  @Override
+  public NodeResource getCurrentNodeResource() {
+    NodeResourceUtils.appendMemoryUnitIfMissing(
+        EngineConnObject.getEngineCreationContext().getOptions());
+
+    CommonNodeResource resource = new CommonNodeResource();
+    LoadResource usedResource = new LoadResource(OverloadUtils.getProcessMaxMemory(), 1);
+    resource.setUsedResource(usedResource);
+    return resource;
+  }
+
+  @Override
+  public String getId() {
+    return Sender.getThisServiceInstance().getInstance() + "_" + id;
+  }
+
+  @Override
+  public int getConcurrentLimit() {
+    return NebulaConfiguration.ENGINE_CONCURRENT_LIMIT.getValue();
+  }
+
+  private NebulaPool getNebulaPool(Map<String, Object> taskParams, Map<String, String> cacheMap) {
+    if (!CollectionUtils.isEmpty(cacheMap)) {
+      configMap.putAll(cacheMap);
+    }
+    taskParams.entrySet().stream()
+        .filter(entry -> entry.getValue() != null)
+        .forEach(entry -> configMap.put(entry.getKey(), String.valueOf(entry.getValue())));
+
+    String host = NebulaConfiguration.NEBULA_HOST.getValue(configMap);
+    Integer port = NebulaConfiguration.NEBULA_PORT.getValue(configMap);
+    Integer maxConnSize = NebulaConfiguration.NEBULA_MAX_CONN_SIZE.getValue(configMap);
+
+    NebulaPool nebulaPool = new NebulaPool();
+    Boolean initResult = false;
+    try {
+
+      NebulaPoolConfig nebulaPoolConfig = new NebulaPoolConfig();
+      nebulaPoolConfig.setMaxConnSize(maxConnSize);
+      List<HostAddress> addresses = Arrays.asList(new HostAddress(host, port));
+      initResult = nebulaPool.init(addresses, nebulaPoolConfig);
+    } catch (Exception e) {
+      logger.error("NebulaPool initialization failed.");
+      throw new NebulaClientException(
+          NebulaErrorCodeSummary.NEBULA_CLIENT_INITIALIZATION_FAILED.getErrorCode(),
+          NebulaErrorCodeSummary.NEBULA_CLIENT_INITIALIZATION_FAILED.getErrorDesc());
+    }
+    if (!initResult) {
+      logger.error("NebulaPool initialization failed.");
+      throw new NebulaClientException(
+          NebulaErrorCodeSummary.NEBULA_CLIENT_INITIALIZATION_FAILED.getErrorCode(),
+          NebulaErrorCodeSummary.NEBULA_CLIENT_INITIALIZATION_FAILED.getErrorDesc());
+    }
+    return nebulaPool;
+  }
+
+  private Session getSession(String taskId, NebulaPool nebulaPool) {
+    if (sessionCache.containsKey(taskId)
+        && sessionCache.get(taskId) != null
+        && sessionCache.get(taskId).ping()) {
+      return sessionCache.get(taskId);
+    } else {
+      Session session;
+      String username = NebulaConfiguration.NEBULA_USER_NAME.getValue(configMap);
+      String password = NebulaConfiguration.NEBULA_PASSWORD.getValue(configMap);
+      Boolean reconnect = NebulaConfiguration.NEBULA_RECONNECT_ENABLED.getValue(configMap);
+
+      try {
+        session = nebulaPool.getSession(username, password, reconnect);
+      } catch (Exception e) {
+        logger.error("Nebula Session initialization failed.");
+        throw new NebulaClientException(
+            NebulaErrorCodeSummary.NEBULA_CLIENT_INITIALIZATION_FAILED.getErrorCode(),
+            NebulaErrorCodeSummary.NEBULA_CLIENT_INITIALIZATION_FAILED.getErrorDesc());
+      }
+
+      sessionCache.put(taskId, session);
+      return session;
+    }
+  }
+
+  private void initialStatusUpdates(
+      String taskId, EngineExecutionContext engineExecutorContext, Session session) {
+    if (session.ping()) {
+      engineExecutorContext.pushProgress(progress(taskId), getProgressInfo(taskId));
+    }
+  }
+
+  private void queryOutput(
+      String taskId, EngineExecutionContext engineExecutorContext, ResultSet resultSet) {
+    int columnCount = 0;
+    ResultSetWriter resultSetWriter =
+        engineExecutorContext.createResultSetWriter(ResultSetFactory.TABLE_TYPE);
+
+    try {
+      List<String> colNames = resultSet.keys();
+
+      if (CollectionUtils.isEmpty(colNames)) {
+        throw new RuntimeException("Nebula columns is null.");
+      }
+
+      List<Column> columns =
+          colNames.stream()
+              .map(column -> new Column(column, DataType.toDataType("string"), ""))
+              .collect(Collectors.toList());
+      columnCount = columns.size();
+      resultSetWriter.addMetaData(new TableMetaData(columns.toArray(new Column[0])));
+      if (!resultSet.isEmpty()) {
+        for (int i = 0; i < resultSet.rowsSize(); i++) {
+          ResultSet.Record record = resultSet.rowValues(i);
+          if (record != null) {
+            String[] rowArray =
+                record.values().stream()
+                    .map(
+                        x -> {
+                          try {
+                            return x.asString();
+                          } catch (Exception e) {
+                            return "";
+                          }
+                        })
+                    .toArray(String[]::new);
+            resultSetWriter.addRecord(new TableRecord(rowArray));
+          }
+        }
+        engineExecutorContext.pushProgress(progress(taskId), getProgressInfo(taskId));
+      }
+    } catch (Exception e) {
+      IOUtils.closeQuietly(resultSetWriter);
+    }
+    String message =
+        String.format("Fetched %d col(s) : %d row(s) in Nebula", columnCount, resultSet.rowsSize());
+    logger.info(message);
+    engineExecutorContext.appendStdout(LogUtils.generateInfo(message));
+    engineExecutorContext.sendResultSet(resultSetWriter);
+  }
+
+  private ErrorExecuteResponse verifyServerError(
+      String taskId, EngineExecutionContext engineExecutorContext, ResultSet resultSet)
+      throws ErrorException {
+    engineExecutorContext.pushProgress(progress(taskId), getProgressInfo(taskId));
+
+    if (!resultSet.isSucceeded() || resultSet.getErrorCode() != ErrorCode.SUCCEEDED.getValue()) {
+      logger.error(
+          "Nebula execute failed (#{}): {}", resultSet.getErrorCode(), resultSet.getErrorMessage());
+      engineExecutorContext.appendStdout(LogUtils.generateERROR(resultSet.getErrorMessage()));
+      return new ErrorExecuteResponse(resultSet.getErrorMessage(), null);
+    }
+    return null;
+  }
+
+  @Override
+  public void killAll() {
+    Iterator<Session> iterator = sessionCache.values().iterator();
+    while (iterator.hasNext()) {
+      Session session = iterator.next();
+      if (session != null) {
+        session.release();
+      }
+    }
+    sessionCache.clear();
+  }
+
+  @Override
+  public void close() {
+    killAll();
+    super.close();
+  }
+}
diff --git a/linkis-engineconn-plugins/nebula/src/main/resources/linkis-engineconn.properties b/linkis-engineconn-plugins/nebula/src/main/resources/linkis-engineconn.properties
new file mode 100644
index 0000000000..059eccb793
--- /dev/null
+++ b/linkis-engineconn-plugins/nebula/src/main/resources/linkis-engineconn.properties
@@ -0,0 +1,23 @@
+#
+# 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.
+#
+
+wds.linkis.server.version=v1
+#wds.linkis.engineconn.debug.enable=true
+#wds.linkis.keytab.enable=true
+wds.linkis.engineconn.plugin.default.class=org.apache.linkis.engineplugin.nebula.NebulaEngineConnPlugin
+
+wds.linkis.engineconn.support.parallelism=true
+
+wds.linkis.engineconn.max.free.time=0
\ No newline at end of file
diff --git a/linkis-engineconn-plugins/nebula/src/main/resources/log4j2.xml b/linkis-engineconn-plugins/nebula/src/main/resources/log4j2.xml
new file mode 100644
index 0000000000..2cd3e264c3
--- /dev/null
+++ b/linkis-engineconn-plugins/nebula/src/main/resources/log4j2.xml
@@ -0,0 +1,91 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ 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.
+  -->
+  
+<configuration status="error" monitorInterval="30">
+    <appenders>
+        <Console name="Console" target="SYSTEM_OUT">
+            <ThresholdFilter level="INFO" onMatch="ACCEPT" onMismatch="DENY"/>
+            <PatternLayout pattern="%d{yyyy-MM-dd HH:mm:ss.SSS} %-5level [%t] %logger{36} %L %M [JobId-%X{jobId}] - %msg%xEx%n"/>
+        </Console>
+
+        <Send name="Send" >
+            <Filters>
+                <ThresholdFilter level="WARN" onMatch="ACCEPT" onMismatch="DENY" />
+            </Filters>
+            <PatternLayout pattern="%d{yyyy-MM-dd HH:mm:ss.SSS} %-5level [%t] %logger{36} %L %M [JobId-%X{jobId}] - %msg%xEx%n"/>
+        </Send>
+
+        <Send name="SendPackage" >
+            <PatternLayout pattern="%d{yyyy-MM-dd HH:mm:ss.SSS} %-5level [%t] %logger{36} %L %M [JobId-%X{jobId}] - %msg%xEx%n"/>
+        </Send>
+
+        <Console name="stderr" target="SYSTEM_ERR">
+            <ThresholdFilter level="ERROR" onMatch="ACCEPT" onMismatch="DENY" />
+            <PatternLayout pattern="%d{HH:mm:ss.SSS} %-5level %class{36} %L %M [JobId-%X{jobId}] - %msg%xEx%n"/>
+        </Console>
+    </appenders>
+
+    <loggers>
+      <root level="INFO">
+            <appender-ref ref="stderr"/>
+            <appender-ref ref="Console"/>
+            <appender-ref ref="Send"/>
+        </root>
+        <logger name="org.apache.hadoop.hive.ql.exec.StatsTask" level="info" additivity="true">
+            <appender-ref ref="SendPackage"/>
+        </logger>
+        <logger name="org.springframework.boot.diagnostics.LoggingFailureAnalysisReporter" level="error" additivity="true">
+            <appender-ref ref="stderr"/>
+        </logger>
+        <logger name="com.netflix.discovery" level="warn" additivity="true">
+            <appender-ref ref="Send"/>
+        </logger>
+        <logger name="org.apache.hadoop.yarn" level="warn" additivity="true">
+            <appender-ref ref="Send"/>
+        </logger>
+        <logger name="org.springframework" level="warn" additivity="true">
+            <appender-ref ref="Send"/>
+        </logger>
+        <logger name="org.apache.linkis.server.security" level="warn" additivity="true">
+            <appender-ref ref="Send"/>
+        </logger>
+        <logger name="org.apache.hadoop.hive.ql.exec.mr.ExecDriver" level="fatal" additivity="true">
+            <appender-ref ref="Send"/>
+        </logger>
+        <logger name="org.apache.hadoop.hdfs.KeyProviderCache" level="fatal" additivity="true">
+            <appender-ref ref="Send"/>
+        </logger>
+        <logger name="org.spark_project.jetty" level="ERROR" additivity="true">
+            <appender-ref ref="Send"/>
+        </logger>
+        <logger name="org.eclipse.jetty" level="ERROR" additivity="true">
+            <appender-ref ref="Send"/>
+        </logger>
+        <logger name="org.springframework" level="ERROR" additivity="true">
+            <appender-ref ref="Send"/>
+        </logger>
+        <logger name="org.reflections.Reflections" level="ERROR" additivity="true">
+            <appender-ref ref="Send"/>
+        </logger>
+
+        <logger name="org.apache.hadoop.ipc.Client" level="ERROR" additivity="true">
+            <appender-ref ref="Send"/>
+        </logger>
+
+   </loggers>
+</configuration>
diff --git a/linkis-engineconn-plugins/nebula/src/main/scala/org/apache/linkis/engineplugin/nebula/factory/NebulaEngineConnFactory.scala b/linkis-engineconn-plugins/nebula/src/main/scala/org/apache/linkis/engineplugin/nebula/factory/NebulaEngineConnFactory.scala
new file mode 100644
index 0000000000..2f7c3c8fb8
--- /dev/null
+++ b/linkis-engineconn-plugins/nebula/src/main/scala/org/apache/linkis/engineplugin/nebula/factory/NebulaEngineConnFactory.scala
@@ -0,0 +1,44 @@
+/*
+ * 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.linkis.engineplugin.nebula.factory
+
+import org.apache.linkis.engineconn.common.creation.EngineCreationContext
+import org.apache.linkis.engineconn.common.engineconn.EngineConn
+import org.apache.linkis.engineconn.computation.executor.creation.ComputationSingleExecutorEngineConnFactory
+import org.apache.linkis.engineconn.executor.entity.LabelExecutor
+import org.apache.linkis.engineplugin.nebula.conf.NebulaConfiguration
+import org.apache.linkis.engineplugin.nebula.executor.NebulaEngineConnExecutor
+import org.apache.linkis.manager.label.entity.engine.{EngineType, RunType}
+import org.apache.linkis.manager.label.entity.engine.EngineType.EngineType
+import org.apache.linkis.manager.label.entity.engine.RunType.RunType
+
+class NebulaEngineConnFactory extends ComputationSingleExecutorEngineConnFactory {
+
+  override def newExecutor(
+      id: Int,
+      engineCreationContext: EngineCreationContext,
+      engineConn: EngineConn
+  ): LabelExecutor = {
+    new NebulaEngineConnExecutor(NebulaConfiguration.ENGINE_DEFAULT_LIMIT.getValue, id)
+  }
+
+  override protected def getEngineConnType: EngineType = EngineType.NEBULA
+
+  override protected def getRunType: RunType = RunType.NEBULA_SQL
+
+}
diff --git a/linkis-engineconn-plugins/spark/src/main/java/org/apache/linkis/engineplugin/spark/client/context/SparkConfig.java b/linkis-engineconn-plugins/spark/src/main/java/org/apache/linkis/engineplugin/spark/client/context/SparkConfig.java
index 24d3ddcb2c..1768b77d04 100644
--- a/linkis-engineconn-plugins/spark/src/main/java/org/apache/linkis/engineplugin/spark/client/context/SparkConfig.java
+++ b/linkis-engineconn-plugins/spark/src/main/java/org/apache/linkis/engineplugin/spark/client/context/SparkConfig.java
@@ -17,6 +17,8 @@
 
 package org.apache.linkis.engineplugin.spark.client.context;
 
+import org.apache.commons.lang3.StringUtils;
+
 import java.util.HashMap;
 import java.util.Map;
 
@@ -48,6 +50,10 @@ public class SparkConfig {
 
   private String k8sNamespace;
   private String k8sFileUploadPath;
+
+  private String k8sDriverRequestCores;
+  private String k8sExecutorRequestCores;
+  private String k8sSparkUIPort;
   private String deployMode = "client"; // ("client") // todo cluster
   private String appResource; // ("")
   private String appName; // ("")
@@ -73,6 +79,7 @@ public class SparkConfig {
   private String principal; // ("--principal", "")
   private String keytab; // ("--keytab", "")
   private String queue; // ("--queue", "")
+  private String pyFiles; // ("--py-files", "")
 
   public String getK8sFileUploadPath() {
     return k8sFileUploadPath;
@@ -131,6 +138,10 @@ public String getK8sConfigFile() {
   }
 
   public void setK8sConfigFile(String k8sConfigFile) {
+    if (StringUtils.isNotBlank(k8sConfigFile) && k8sConfigFile.startsWith("~")) {
+      String user = System.getProperty("user.home");
+      k8sConfigFile = k8sConfigFile.replaceFirst("~", user);
+    }
     this.k8sConfigFile = k8sConfigFile;
   }
 
@@ -170,6 +181,30 @@ public void setK8sImage(String k8sImage) {
     this.k8sImage = k8sImage;
   }
 
+  public String getK8sDriverRequestCores() {
+    return k8sDriverRequestCores;
+  }
+
+  public void setK8sDriverRequestCores(String k8sDriverRequestCores) {
+    this.k8sDriverRequestCores = k8sDriverRequestCores;
+  }
+
+  public String getK8sExecutorRequestCores() {
+    return k8sExecutorRequestCores;
+  }
+
+  public void setK8sExecutorRequestCores(String k8sExecutorRequestCores) {
+    this.k8sExecutorRequestCores = k8sExecutorRequestCores;
+  }
+
+  public String getK8sSparkUIPort() {
+    return k8sSparkUIPort;
+  }
+
+  public void setK8sSparkUIPort(String k8sSparkUIPort) {
+    this.k8sSparkUIPort = k8sSparkUIPort;
+  }
+
   public String getJavaHome() {
     return javaHome;
   }
@@ -394,6 +429,14 @@ public void setQueue(String queue) {
     this.queue = queue;
   }
 
+  public String getPyFiles() {
+    return pyFiles;
+  }
+
+  public void setPyFiles(String pyFiles) {
+    this.pyFiles = pyFiles;
+  }
+
   @Override
   public String toString() {
     return "SparkConfig{"
@@ -436,6 +479,15 @@ public String toString() {
         + ", k8sNamespace='"
         + k8sNamespace
         + '\''
+        + ", k8sDriverRequestCores='"
+        + k8sDriverRequestCores
+        + '\''
+        + ", k8sExecutorRequestCores='"
+        + k8sExecutorRequestCores
+        + '\''
+        + ", k8sSparkUIPort='"
+        + k8sSparkUIPort
+        + '\''
         + ", deployMode='"
         + deployMode
         + '\''
@@ -503,6 +555,9 @@ public String toString() {
         + ", queue='"
         + queue
         + '\''
+        + ", pyFiles='"
+        + pyFiles
+        + '\''
         + '}';
   }
 }
diff --git a/linkis-engineconn-plugins/spark/src/main/java/org/apache/linkis/engineplugin/spark/client/deployment/ClusterDescriptorAdapterFactory.java b/linkis-engineconn-plugins/spark/src/main/java/org/apache/linkis/engineplugin/spark/client/deployment/ClusterDescriptorAdapterFactory.java
index 91d3eafb6f..bc67a33e9f 100644
--- a/linkis-engineconn-plugins/spark/src/main/java/org/apache/linkis/engineplugin/spark/client/deployment/ClusterDescriptorAdapterFactory.java
+++ b/linkis-engineconn-plugins/spark/src/main/java/org/apache/linkis/engineplugin/spark/client/deployment/ClusterDescriptorAdapterFactory.java
@@ -29,8 +29,13 @@ public static ClusterDescriptorAdapter create(ExecutionContext executionContext)
     ClusterDescriptorAdapter clusterDescriptorAdapter =
         new YarnApplicationClusterDescriptorAdapter(executionContext);
 
-    if (StringUtils.isNotBlank(master) && master.equalsIgnoreCase("k8s-operator")) {
-      clusterDescriptorAdapter = new KubernetesOperatorClusterDescriptorAdapter(executionContext);
+    if (StringUtils.isNotBlank(master)) {
+      if (master.equalsIgnoreCase("k8s-operator")) {
+        clusterDescriptorAdapter = new KubernetesOperatorClusterDescriptorAdapter(executionContext);
+      } else if (master.equalsIgnoreCase("k8s-native")) {
+        clusterDescriptorAdapter =
+            new KubernetesApplicationClusterDescriptorAdapter(executionContext);
+      }
     }
 
     return clusterDescriptorAdapter;
diff --git a/linkis-engineconn-plugins/spark/src/main/java/org/apache/linkis/engineplugin/spark/client/deployment/KubernetesApplicationClusterDescriptorAdapter.java b/linkis-engineconn-plugins/spark/src/main/java/org/apache/linkis/engineplugin/spark/client/deployment/KubernetesApplicationClusterDescriptorAdapter.java
new file mode 100644
index 0000000000..ce709b2e7a
--- /dev/null
+++ b/linkis-engineconn-plugins/spark/src/main/java/org/apache/linkis/engineplugin/spark/client/deployment/KubernetesApplicationClusterDescriptorAdapter.java
@@ -0,0 +1,233 @@
+/*
+ * 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.linkis.engineplugin.spark.client.deployment;
+
+import org.apache.linkis.engineplugin.spark.client.context.ExecutionContext;
+import org.apache.linkis.engineplugin.spark.client.context.SparkConfig;
+import org.apache.linkis.engineplugin.spark.client.deployment.util.KubernetesHelper;
+
+import org.apache.commons.lang3.StringUtils;
+import org.apache.logging.log4j.util.Strings;
+import org.apache.spark.launcher.CustomSparkSubmitLauncher;
+import org.apache.spark.launcher.SparkAppHandle;
+import org.apache.spark.launcher.SparkLauncher;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Map;
+import java.util.UUID;
+
+import io.fabric8.kubernetes.api.model.Pod;
+import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.dsl.PodResource;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class KubernetesApplicationClusterDescriptorAdapter extends ClusterDescriptorAdapter {
+  private static final Logger logger =
+      LoggerFactory.getLogger(KubernetesOperatorClusterDescriptorAdapter.class);
+
+  protected SparkConfig sparkConfig;
+  protected KubernetesClient client;
+  protected String driverPodName;
+  protected String namespace;
+
+  public KubernetesApplicationClusterDescriptorAdapter(ExecutionContext executionContext) {
+    super(executionContext);
+    this.sparkConfig = executionContext.getSparkConfig();
+    this.client =
+        KubernetesHelper.getKubernetesClient(
+            this.sparkConfig.getK8sConfigFile(),
+            this.sparkConfig.getK8sMasterUrl(),
+            this.sparkConfig.getK8sUsername(),
+            this.sparkConfig.getK8sPassword());
+  }
+
+  public void deployCluster(String mainClass, String args, Map<String, String> confMap)
+      throws IOException {
+    SparkConfig sparkConfig = executionContext.getSparkConfig();
+    sparkLauncher = new CustomSparkSubmitLauncher();
+    sparkLauncher
+        .setJavaHome(sparkConfig.getJavaHome())
+        .setSparkHome(sparkConfig.getSparkHome())
+        .setMaster(sparkConfig.getK8sMasterUrl())
+        .setDeployMode(sparkConfig.getDeployMode())
+        .setAppName(sparkConfig.getAppName())
+        .setVerbose(true);
+    this.driverPodName = generateDriverPodName(sparkConfig.getAppName());
+    this.namespace = sparkConfig.getK8sNamespace();
+    setConf(sparkLauncher, "spark.app.name", sparkConfig.getAppName());
+    setConf(sparkLauncher, "spark.ui.port", sparkConfig.getK8sSparkUIPort());
+    setConf(sparkLauncher, "spark.kubernetes.namespace", this.namespace);
+    setConf(sparkLauncher, "spark.kubernetes.container.image", sparkConfig.getK8sImage());
+    setConf(sparkLauncher, "spark.kubernetes.driver.pod.name", this.driverPodName);
+    setConf(
+        sparkLauncher,
+        "spark.kubernetes.driver.request.cores",
+        sparkConfig.getK8sDriverRequestCores());
+    setConf(
+        sparkLauncher,
+        "spark.kubernetes.executor.request.cores",
+        sparkConfig.getK8sExecutorRequestCores());
+    setConf(
+        sparkLauncher,
+        "spark.kubernetes.container.image.pullPolicy",
+        sparkConfig.getK8sImagePullPolicy());
+    setConf(
+        sparkLauncher,
+        "spark.kubernetes.authenticate.driver.serviceAccountName",
+        sparkConfig.getK8sServiceAccount());
+    if (confMap != null) confMap.forEach((k, v) -> sparkLauncher.setConf(k, v));
+
+    addSparkArg(sparkLauncher, "--jars", sparkConfig.getJars());
+    addSparkArg(sparkLauncher, "--packages", sparkConfig.getPackages());
+    addSparkArg(sparkLauncher, "--exclude-packages", sparkConfig.getExcludePackages());
+    addSparkArg(sparkLauncher, "--repositories", sparkConfig.getRepositories());
+    addSparkArg(sparkLauncher, "--files", sparkConfig.getFiles());
+    addSparkArg(sparkLauncher, "--archives", sparkConfig.getArchives());
+    addSparkArg(sparkLauncher, "--driver-memory", sparkConfig.getDriverMemory());
+    addSparkArg(sparkLauncher, "--driver-java-options", sparkConfig.getDriverJavaOptions());
+    addSparkArg(sparkLauncher, "--driver-library-path", sparkConfig.getDriverLibraryPath());
+    addSparkArg(sparkLauncher, "--driver-class-path", sparkConfig.getDriverClassPath());
+    addSparkArg(sparkLauncher, "--executor-memory", sparkConfig.getExecutorMemory());
+    addSparkArg(sparkLauncher, "--proxy-user", sparkConfig.getProxyUser());
+    addSparkArg(sparkLauncher, "--driver-cores", sparkConfig.getDriverCores().toString());
+    addSparkArg(sparkLauncher, "--total-executor-cores", sparkConfig.getTotalExecutorCores());
+    addSparkArg(sparkLauncher, "--executor-cores", sparkConfig.getExecutorCores().toString());
+    addSparkArg(sparkLauncher, "--num-executors", sparkConfig.getNumExecutors().toString());
+    addSparkArg(sparkLauncher, "--principal", sparkConfig.getPrincipal());
+    addSparkArg(sparkLauncher, "--keytab", sparkConfig.getKeytab());
+    addSparkArg(sparkLauncher, "--py-files", sparkConfig.getPyFiles());
+    sparkLauncher.setAppResource(sparkConfig.getAppResource());
+    sparkLauncher.setMainClass(mainClass);
+    Arrays.stream(args.split("\\s+"))
+        .filter(StringUtils::isNotBlank)
+        .forEach(arg -> sparkLauncher.addAppArgs(arg));
+    sparkAppHandle =
+        sparkLauncher.startApplication(
+            new SparkAppHandle.Listener() {
+              @Override
+              public void stateChanged(SparkAppHandle sparkAppHandle) {}
+
+              @Override
+              public void infoChanged(SparkAppHandle sparkAppHandle) {}
+            });
+    sparkLauncher.setSparkAppHandle(sparkAppHandle);
+  }
+
+  private void addSparkArg(SparkLauncher sparkLauncher, String key, String value) {
+    if (StringUtils.isNotBlank(key) && StringUtils.isNotBlank(value)) {
+      sparkLauncher.addSparkArg(key, value);
+    }
+  }
+
+  private void setConf(SparkLauncher sparkLauncher, String key, String value) {
+    if (StringUtils.isNotBlank(key) && StringUtils.isNotBlank(value)) {
+      sparkLauncher.setConf(key, value);
+    }
+  }
+
+  public boolean initJobId() {
+    Pod sparkDriverPod = getSparkDriverPod();
+    if (null == sparkDriverPod) {
+      return false;
+    }
+    String sparkDriverPodPhase = sparkDriverPod.getStatus().getPhase();
+    String sparkApplicationId = sparkDriverPod.getMetadata().getLabels().get("spark-app-selector");
+
+    if (Strings.isNotBlank(sparkApplicationId)) {
+      this.applicationId = sparkApplicationId;
+    }
+    if (Strings.isNotBlank(sparkDriverPodPhase)) {
+      this.jobState = kubernetesPodStateConvertSparkState(sparkDriverPodPhase);
+    }
+
+    // When the job is not finished, the appId is monitored; otherwise, the status is
+    // monitored(当任务没结束时，监控appId，反之，则监控状态，这里主要防止任务过早结束，导致一直等待)
+    return null != getApplicationId() || (jobState != null && jobState.isFinal());
+  }
+
+  protected Pod getSparkDriverPod() {
+    return client.pods().inNamespace(namespace).withName(driverPodName).get();
+  }
+
+  public String getSparkUIUrl() {
+    Pod sparkDriverPod = getSparkDriverPod();
+    if (null != sparkDriverPod) {
+      String sparkDriverPodIP = sparkDriverPod.getStatus().getPodIP();
+      if (StringUtils.isNotBlank(sparkDriverPodIP)) {
+        return sparkDriverPodIP + ":" + this.sparkConfig.getK8sSparkUIPort();
+      } else {
+        logger.info("spark driver pod IP is null, the application may be pending");
+      }
+    } else {
+      logger.info("spark driver pod is not exist");
+    }
+    return "";
+  }
+
+  @Override
+  public SparkAppHandle.State getJobState() {
+    Pod sparkDriverPod = getSparkDriverPod();
+    if (null != sparkDriverPod) {
+      String sparkDriverPodPhase = sparkDriverPod.getStatus().getPhase();
+      this.jobState = kubernetesPodStateConvertSparkState(sparkDriverPodPhase);
+      logger.info("Job {} state is {}.", getApplicationId(), this.jobState);
+      return this.jobState;
+    }
+    return null;
+  }
+
+  @Override
+  public void close() {
+    logger.info("Start to close job {}.", getApplicationId());
+    PodResource<Pod> sparkDriverPodResource =
+        client.pods().inNamespace(namespace).withName(driverPodName);
+    if (null != sparkDriverPodResource.get()) {
+      sparkDriverPodResource.delete();
+    }
+    client.close();
+  }
+
+  @Override
+  public boolean isDisposed() {
+    return this.jobState.isFinal();
+  }
+
+  public SparkAppHandle.State kubernetesPodStateConvertSparkState(String kubernetesState) {
+    if (StringUtils.isBlank(kubernetesState)) {
+      return SparkAppHandle.State.UNKNOWN;
+    }
+    switch (kubernetesState.toUpperCase()) {
+      case "PENDING":
+        return SparkAppHandle.State.CONNECTED;
+      case "RUNNING":
+        return SparkAppHandle.State.RUNNING;
+      case "SUCCEEDED":
+        return SparkAppHandle.State.FINISHED;
+      case "FAILED":
+        return SparkAppHandle.State.FAILED;
+      default:
+        return SparkAppHandle.State.UNKNOWN;
+    }
+  }
+
+  public String generateDriverPodName(String appName) {
+    return appName + "-" + UUID.randomUUID().toString().replace("-", "") + "-driver";
+  }
+}
diff --git a/linkis-engineconn-plugins/spark/src/main/java/org/apache/linkis/engineplugin/spark/client/deployment/KubernetesOperatorClusterDescriptorAdapter.java b/linkis-engineconn-plugins/spark/src/main/java/org/apache/linkis/engineplugin/spark/client/deployment/KubernetesOperatorClusterDescriptorAdapter.java
index fa6236600e..eafa8abec5 100644
--- a/linkis-engineconn-plugins/spark/src/main/java/org/apache/linkis/engineplugin/spark/client/deployment/KubernetesOperatorClusterDescriptorAdapter.java
+++ b/linkis-engineconn-plugins/spark/src/main/java/org/apache/linkis/engineplugin/spark/client/deployment/KubernetesOperatorClusterDescriptorAdapter.java
@@ -38,6 +38,9 @@
 import io.fabric8.kubernetes.api.model.apiextensions.v1.CustomResourceDefinitionList;
 import io.fabric8.kubernetes.client.CustomResource;
 import io.fabric8.kubernetes.client.KubernetesClient;
+import io.fabric8.kubernetes.client.Watcher;
+import io.fabric8.kubernetes.client.WatcherException;
+import io.fabric8.kubernetes.client.dsl.MixedOperation;
 import io.fabric8.kubernetes.client.dsl.NonNamespaceOperation;
 import io.fabric8.kubernetes.client.dsl.Resource;
 import org.slf4j.Logger;
@@ -152,32 +155,46 @@ public void deployCluster(String mainClass, String args, Map<String, String> con
   }
 
   public boolean initJobId() {
-    SparkApplicationStatus sparkApplicationStatus = getKubernetesOperatorState();
-
-    if (Objects.nonNull(sparkApplicationStatus)) {
-      this.applicationId = sparkApplicationStatus.getSparkApplicationId();
-      this.jobState =
-          kubernetesOperatorStateConvertSparkState(
-              sparkApplicationStatus.getApplicationState().getState());
+    try {
+      getKubernetesOperatorState();
+    } catch (Exception e) {
+      try {
+        // Prevent watch interruption due to network interruption.Restart Watcher.
+        Thread.sleep(5000);
+        getKubernetesOperatorState();
+      } catch (InterruptedException interruptedException) {
+        logger.error("Use k8s watch obtain the status failed");
+      }
     }
-
     // When the job is not finished, the appId is monitored; otherwise, the status is
     // monitored(当任务没结束时，监控appId，反之，则监控状态，这里主要防止任务过早结束，导致一直等待)
     return null != getApplicationId() || (jobState != null && jobState.isFinal());
   }
 
-  private SparkApplicationStatus getKubernetesOperatorState() {
-    List<SparkApplication> sparkApplicationList =
-        getSparkApplicationClient(client).list().getItems();
-    if (CollectionUtils.isNotEmpty(sparkApplicationList)) {
-      for (SparkApplication sparkApplication : sparkApplicationList) {
-        if (sparkApplication.getMetadata().getNamespace().equals(this.sparkConfig.getK8sNamespace())
-            && sparkApplication.getMetadata().getName().equals(this.sparkConfig.getAppName())) {
-          return sparkApplication.getStatus();
-        }
-      }
-    }
-    return null;
+  private void getKubernetesOperatorState() {
+    getSparkApplicationClient(client)
+        .inNamespace(this.sparkConfig.getK8sNamespace())
+        .withName(this.sparkConfig.getAppName())
+        .watch(
+            new Watcher<SparkApplication>() {
+              @Override
+              public void eventReceived(Action action, SparkApplication sparkApplication) {
+                // todo get status
+                if (Objects.nonNull(sparkApplication.getStatus())) {
+                  applicationId = sparkApplication.getStatus().getSparkApplicationId();
+                  jobState =
+                      kubernetesOperatorStateConvertSparkState(
+                          sparkApplication.getStatus().getApplicationState().getState());
+                }
+              }
+
+              @Override
+              public void onClose(WatcherException e) {
+                // Invoked when the watcher closes due to an Exception.Restart Watcher.
+                logger.error("Use k8s watch obtain the status failed", e);
+                getKubernetesOperatorState();
+              }
+            });
   }
 
   public SparkAppHandle.State kubernetesOperatorStateConvertSparkState(String kubernetesState) {
@@ -216,8 +233,7 @@ public void close() {
     client.close();
   }
 
-  public static NonNamespaceOperation<
-          SparkApplication, SparkApplicationList, Resource<SparkApplication>>
+  public static MixedOperation<SparkApplication, SparkApplicationList, Resource<SparkApplication>>
       getSparkApplicationClient(KubernetesClient client) {
     return client.customResources(SparkApplication.class, SparkApplicationList.class);
   }
diff --git a/linkis-engineconn-plugins/spark/src/main/resources/spark-k8s-operator.md b/linkis-engineconn-plugins/spark/src/main/resources/spark-k8s-operator.md
index cabdd2245a..9a0921f299 100644
--- a/linkis-engineconn-plugins/spark/src/main/resources/spark-k8s-operator.md
+++ b/linkis-engineconn-plugins/spark/src/main/resources/spark-k8s-operator.md
@@ -48,8 +48,6 @@ POST /api/rest_j/v1/entrance/submit
 ```json
 {
   "executionContent": {
-    "spark.app.main.class": "org.apache.spark.examples.SparkPi",
-    "spark.app.args": "spark.app.args",
     "runType": "jar",
     "code": "show databases"
   },
@@ -61,13 +59,13 @@ POST /api/rest_j/v1/entrance/submit
         "spark.executor.memory": "1g",
         "spark.driver.memory": "1g",
         "spark.executor.cores": "1",
+        "spark.app.main.class": "org.apache.spark.examples.SparkPi",
         "spark.app.name": "spark-submit-jar-cjtest",
         "spark.app.resource": "local:///opt/spark/examples/jars/spark-examples_2.12-3.2.1.jar",
         "spark.executor.instances": 1,
-        "spark.master": "k8soperator",
-        "spark.k8s.master.url": "http://ip:port",
-        "spark.k8s.username": "username",
-        "spark.k8s.password": "password"
+        "spark.master":"k8s-operator",
+        "linkis.spark.k8s.config.file":"~/.kube/config",
+        "linkis.spark.k8s.serviceAccount":"spark"
       }
     }
   },
diff --git a/linkis-engineconn-plugins/spark/src/main/scala/org/apache/linkis/engineplugin/spark/config/SparkConfiguration.scala b/linkis-engineconn-plugins/spark/src/main/scala/org/apache/linkis/engineplugin/spark/config/SparkConfiguration.scala
index b42fc0934e..bb079b7b56 100644
--- a/linkis-engineconn-plugins/spark/src/main/scala/org/apache/linkis/engineplugin/spark/config/SparkConfiguration.scala
+++ b/linkis-engineconn-plugins/spark/src/main/scala/org/apache/linkis/engineplugin/spark/config/SparkConfiguration.scala
@@ -68,6 +68,7 @@ object SparkConfiguration extends Logging {
   val SPARK_K8S_RESTART_POLICY = CommonVars[String]("linkis.spark.k8s.restartPolicy", "Never")
   val SPARK_K8S_SPARK_VERSION = CommonVars[String]("linkis.spark.k8s.sparkVersion", "3.2.1")
   val SPARK_K8S_NAMESPACE = CommonVars[String]("linkis.spark.k8s.namespace", "default")
+  val SPARK_K8S_UI_PORT = CommonVars[String]("linkis.spark.k8s.ui.port", "4040")
 
   val SPARK_K8S_EXECUTOR_REQUEST_CORES =
     CommonVars[String]("linkis.spark.k8s.executor.request.cores", "1")
@@ -80,6 +81,8 @@ object SparkConfiguration extends Logging {
 
   val SPARK_PYTHON_VERSION = CommonVars[String]("spark.python.version", "python")
 
+  val SPARK_PYTHON_FILES = CommonVars[String]("spark.submit.pyFiles", "")
+
   val SPARK_PYTHON_TEST_MODE_ENABLE =
     CommonVars[Boolean]("linkis.spark.python.test.mode.enable", false)
 
diff --git a/linkis-engineconn-plugins/spark/src/main/scala/org/apache/linkis/engineplugin/spark/executor/SparkOnKubernetesSubmitOnceExecutor.scala b/linkis-engineconn-plugins/spark/src/main/scala/org/apache/linkis/engineplugin/spark/executor/SparkOnKubernetesSubmitOnceExecutor.scala
new file mode 100644
index 0000000000..e8b7dfb489
--- /dev/null
+++ b/linkis-engineconn-plugins/spark/src/main/scala/org/apache/linkis/engineplugin/spark/executor/SparkOnKubernetesSubmitOnceExecutor.scala
@@ -0,0 +1,164 @@
+/*
+ * 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.linkis.engineplugin.spark.executor
+
+import org.apache.linkis.common.utils.{ByteTimeUtils, Utils}
+import org.apache.linkis.engineconn.once.executor.{
+  OnceExecutorExecutionContext,
+  OperableOnceExecutor
+}
+import org.apache.linkis.engineplugin.spark.client.deployment.{
+  KubernetesApplicationClusterDescriptorAdapter,
+  YarnApplicationClusterDescriptorAdapter
+}
+import org.apache.linkis.engineplugin.spark.config.SparkConfiguration.{
+  SPARK_APP_CONF,
+  SPARK_APPLICATION_ARGS,
+  SPARK_APPLICATION_MAIN_CLASS
+}
+import org.apache.linkis.engineplugin.spark.context.SparkEngineConnContext
+import org.apache.linkis.engineplugin.spark.utils.SparkJobProgressUtil
+import org.apache.linkis.manager.common.entity.resource._
+import org.apache.linkis.manager.common.utils.ResourceUtils
+import org.apache.linkis.protocol.engine.JobProgressInfo
+
+import org.apache.commons.lang3.StringUtils
+
+import java.util
+
+import scala.concurrent.duration.Duration
+
+import io.fabric8.kubernetes.api.model.Quantity
+
+class SparkOnKubernetesSubmitOnceExecutor(
+    override val id: Long,
+    override protected val sparkEngineConnContext: SparkEngineConnContext
+) extends SparkOnceExecutor[KubernetesApplicationClusterDescriptorAdapter]
+    with OperableOnceExecutor {
+
+  private var oldProgress: Float = 0f
+
+  override def doSubmit(
+      onceExecutorExecutionContext: OnceExecutorExecutionContext,
+      options: Map[String, String]
+  ): Unit = {
+    val args = SPARK_APPLICATION_ARGS.getValue(options)
+    val mainClass = SPARK_APPLICATION_MAIN_CLASS.getValue(options)
+    val extConf = SPARK_APP_CONF.getValue(options)
+    val confMap = new util.HashMap[String, String]()
+    if (StringUtils.isNotBlank(extConf)) {
+      for (conf <- extConf.split("\n")) {
+        if (StringUtils.isNotBlank(conf)) {
+          val pair = conf.trim.split("=")
+          if (pair.length == 2) {
+            confMap.put(pair(0), pair(1))
+          } else {
+            logger.warn(s"ignore spark conf: $conf")
+          }
+        }
+      }
+    }
+    logger.info(
+      s"Ready to submit spark application to kubernetes, mainClass: $mainClass, args: $args."
+    )
+    clusterDescriptorAdapter.deployCluster(mainClass, args, confMap)
+  }
+
+  override protected def waitToRunning(): Unit = {
+    // Wait until the task return applicationId (等待返回applicationId)
+    Utils.waitUntil(() => clusterDescriptorAdapter.initJobId(), Duration.Inf)
+    // Synchronize applicationId to EC SparkOnceExecutor to facilitate user operations,
+    // such as obtaining progress and killing jobs(将applicationId同步给EC执行器，方便用户操作，如获取进度，kill任务等)
+    setApplicationId(clusterDescriptorAdapter.getApplicationId)
+    super.waitToRunning()
+  }
+
+  override def getApplicationURL: String = ""
+
+  override def getCurrentNodeResource(): NodeResource = {
+    logger.info("Begin to get actual used resources!")
+    Utils.tryCatch({
+      val sparkConf = sparkEngineConnContext.getExecutionContext.getSparkConfig
+      val sparkNamespace = sparkConf.getK8sNamespace
+
+      val executorNum: Int = sparkConf.getNumExecutors
+      val executorMem: Long =
+        ByteTimeUtils.byteStringAsBytes(sparkConf.getExecutorMemory) * executorNum
+      val driverMem: Long = ByteTimeUtils.byteStringAsBytes(sparkConf.getDriverMemory)
+
+      val executorCoresQuantity = Quantity.parse(sparkConf.getK8sExecutorRequestCores)
+      val executorCores: Long =
+        (Quantity.getAmountInBytes(executorCoresQuantity).doubleValue() * 1000).toLong * executorNum
+      val driverCoresQuantity = Quantity.parse(sparkConf.getK8sDriverRequestCores)
+      val driverCores: Long =
+        (Quantity.getAmountInBytes(driverCoresQuantity).doubleValue() * 1000).toLong
+
+      logger.info(
+        "Current actual used resources is driverMem:" + driverMem + ",driverCores:" + driverCores + ",executorMem:" + executorMem + ",executorCores:" + executorCores + ",namespace:" + sparkNamespace
+      )
+      val usedResource = new DriverAndKubernetesResource(
+        new LoadInstanceResource(0, 0, 0),
+        new KubernetesResource(executorMem + driverMem, executorCores + driverCores, sparkNamespace)
+      )
+      val nodeResource = new CommonNodeResource
+      nodeResource.setUsedResource(usedResource)
+      nodeResource.setResourceType(ResourceUtils.getResourceTypeByResource(usedResource))
+      nodeResource
+    })(t => {
+      logger.warn("Get actual used resource exception", t)
+      null
+    })
+  }
+
+  override def getProgress: Float = {
+    val jobIsFinal = clusterDescriptorAdapter != null &&
+      clusterDescriptorAdapter.getJobState != null &&
+      clusterDescriptorAdapter.getJobState.isFinal
+    if (oldProgress >= 1 || jobIsFinal) {
+      1
+    } else {
+      val sparkUIUrl = this.clusterDescriptorAdapter.getSparkUIUrl
+      if (StringUtils.isNotBlank(sparkUIUrl)) {
+        val newProgress =
+          SparkJobProgressUtil.getProgress(this.getApplicationId, sparkUIUrl)
+        if (newProgress > oldProgress) {
+          oldProgress = newProgress
+        }
+      }
+      oldProgress
+    }
+  }
+
+  override def getProgressInfo: Array[JobProgressInfo] = {
+    val sparkUIUrl = this.clusterDescriptorAdapter.getSparkUIUrl
+    if (StringUtils.isNotBlank(sparkUIUrl)) {
+      SparkJobProgressUtil.getSparkJobProgressInfo(this.getApplicationId, sparkUIUrl)
+    } else {
+      Array.empty
+    }
+  }
+
+  override def getMetrics: util.Map[String, Any] = {
+    new util.HashMap[String, Any]()
+  }
+
+  override def getDiagnosis: util.Map[String, Any] = {
+    new util.HashMap[String, Any]()
+  }
+
+}
diff --git a/linkis-engineconn-plugins/spark/src/main/scala/org/apache/linkis/engineplugin/spark/factory/SparkEngineConnFactory.scala b/linkis-engineconn-plugins/spark/src/main/scala/org/apache/linkis/engineplugin/spark/factory/SparkEngineConnFactory.scala
index bc18e2badf..a237b544b4 100644
--- a/linkis-engineconn-plugins/spark/src/main/scala/org/apache/linkis/engineplugin/spark/factory/SparkEngineConnFactory.scala
+++ b/linkis-engineconn-plugins/spark/src/main/scala/org/apache/linkis/engineplugin/spark/factory/SparkEngineConnFactory.scala
@@ -39,7 +39,6 @@ import org.apache.linkis.manager.engineplugin.common.creation.{
 }
 import org.apache.linkis.manager.engineplugin.common.launch.process.Environment
 import org.apache.linkis.manager.engineplugin.common.launch.process.Environment.variable
-import org.apache.linkis.manager.label.constant.LabelValueConstant
 import org.apache.linkis.manager.label.entity.engine.EngineType
 import org.apache.linkis.manager.label.entity.engine.EngineType.EngineType
 import org.apache.linkis.manager.label.utils.LabelUtil
@@ -86,12 +85,13 @@ class SparkEngineConnFactory extends MultiExecutorEngineConnFactory with Logging
     val hadoopConfDir = EnvConfiguration.HADOOP_CONF_DIR.getValue(options)
     val sparkHome = SPARK_HOME.getValue(options)
     val sparkConfDir = SPARK_CONF_DIR.getValue(options)
-    val sparkConfig: SparkConfig = getSparkConfig(options)
+    val sparkConfig: SparkConfig =
+      getSparkConfig(options, LabelUtil.isYarnClusterMode(engineCreationContext.getLabels()))
     val context = new EnvironmentContext(sparkConfig, hadoopConfDir, sparkConfDir, sparkHome, null)
     context
   }
 
-  def getSparkConfig(options: util.Map[String, String]): SparkConfig = {
+  def getSparkConfig(options: util.Map[String, String], isYarnClusterMode: Boolean): SparkConfig = {
     logger.info("options: " + JsonUtils.jackson.writeValueAsString(options))
     val sparkConfig: SparkConfig = new SparkConfig()
     sparkConfig.setJavaHome(variable(Environment.JAVA_HOME))
@@ -111,8 +111,18 @@ class SparkEngineConnFactory extends MultiExecutorEngineConnFactory with Logging
       sparkConfig.setK8sRestartPolicy(SPARK_K8S_RESTART_POLICY.getValue(options))
       sparkConfig.setK8sLanguageType(SPARK_K8S_LANGUAGE_TYPE.getValue(options))
       sparkConfig.setK8sImagePullPolicy(SPARK_K8S_IMAGE_PULL_POLICY.getValue(options))
+      sparkConfig.setK8sDriverRequestCores(SPARK_K8S_DRIVER_REQUEST_CORES.getValue(options))
+      sparkConfig.setK8sExecutorRequestCores(SPARK_K8S_EXECUTOR_REQUEST_CORES.getValue(options))
+      sparkConfig.setK8sSparkUIPort(SPARK_K8S_UI_PORT.getValue(options))
+    }
+
+    if (master.startsWith("yarn")) {
+      if (isYarnClusterMode) {
+        sparkConfig.setDeployMode(SparkConfiguration.SPARK_YARN_CLUSTER)
+      } else {
+        sparkConfig.setDeployMode(SparkConfiguration.SPARK_YARN_CLIENT)
+      }
     }
-    sparkConfig.setDeployMode(SPARK_DEPLOY_MODE.getValue(options))
     sparkConfig.setAppResource(SPARK_APP_RESOURCE.getValue(options))
     sparkConfig.setAppName(SPARK_APP_NAME.getValue(options))
     sparkConfig.setJars(SPARK_EXTRA_JARS.getValue(options)) // todo
@@ -125,6 +135,7 @@ class SparkEngineConnFactory extends MultiExecutorEngineConnFactory with Logging
     sparkConfig.setExecutorCores(LINKIS_SPARK_EXECUTOR_CORES.getValue(options))
     sparkConfig.setNumExecutors(LINKIS_SPARK_EXECUTOR_INSTANCES.getValue(options))
     sparkConfig.setQueue(LINKIS_QUEUE_NAME.getValue(options))
+    sparkConfig.setPyFiles(SPARK_PYTHON_FILES.getValue(options))
 
     logger.info(s"spark_info: ${sparkConfig}")
     sparkConfig
@@ -147,10 +158,7 @@ class SparkEngineConnFactory extends MultiExecutorEngineConnFactory with Logging
       sparkConf.getOption("spark.master").getOrElse(CommonVars("spark.master", "yarn").getValue)
     logger.info(s"------ Create new SparkContext {$master} -------")
 
-    val label = LabelUtil.getEngingeConnRuntimeModeLabel(engineCreationContext.getLabels())
-    val isYarnClusterMode: Boolean =
-      if (null != label && label.getModeValue.equals(LabelValueConstant.YARN_CLUSTER_VALUE)) true
-      else false
+    val isYarnClusterMode = LabelUtil.isYarnClusterMode(engineCreationContext.getLabels())
 
     if (isYarnClusterMode) {
       sparkConf.set("spark.submit.deployMode", "cluster")
diff --git a/linkis-engineconn-plugins/spark/src/main/scala/org/apache/linkis/engineplugin/spark/factory/SparkEngineConnResourceFactory.scala b/linkis-engineconn-plugins/spark/src/main/scala/org/apache/linkis/engineplugin/spark/factory/SparkEngineConnResourceFactory.scala
index 922826c2ab..640476a589 100644
--- a/linkis-engineconn-plugins/spark/src/main/scala/org/apache/linkis/engineplugin/spark/factory/SparkEngineConnResourceFactory.scala
+++ b/linkis-engineconn-plugins/spark/src/main/scala/org/apache/linkis/engineplugin/spark/factory/SparkEngineConnResourceFactory.scala
@@ -113,7 +113,9 @@ class SparkEngineConnResourceFactory extends AbstractEngineResourceFactory with
         Quantity.parse(SPARK_K8S_EXECUTOR_REQUEST_CORES.getValue(properties))
       (Quantity.getAmountInBytes(executorCoresQuantity).doubleValue() * 1000).toLong
     } else {
-      LINKIS_SPARK_EXECUTOR_CORES.getValue(properties) * 1000L
+      val sparkDefaultExecutorCores: Int = LINKIS_SPARK_EXECUTOR_CORES.getValue(properties)
+      properties.put(SPARK_K8S_EXECUTOR_REQUEST_CORES.key, sparkDefaultExecutorCores.toString)
+      sparkDefaultExecutorCores * 1000L
     }
     val executorMemory = LINKIS_SPARK_EXECUTOR_MEMORY.getValue(properties)
     val executorMemoryWithUnit = if (StringUtils.isNumeric(executorMemory)) {
@@ -126,7 +128,9 @@ class SparkEngineConnResourceFactory extends AbstractEngineResourceFactory with
         Quantity.parse(SPARK_K8S_DRIVER_REQUEST_CORES.getValue(properties))
       (Quantity.getAmountInBytes(executorCoresQuantity).doubleValue() * 1000).toLong
     } else {
-      LINKIS_SPARK_DRIVER_CORES.getValue(properties) * 1000L
+      val sparkDefaultDriverCores: Int = LINKIS_SPARK_DRIVER_CORES.getValue(properties)
+      properties.put(SPARK_K8S_DRIVER_REQUEST_CORES.key, sparkDefaultDriverCores.toString)
+      sparkDefaultDriverCores * 1000L
     }
     val driverMemory = LINKIS_SPARK_DRIVER_MEMORY.getValue(properties)
     val driverMemoryWithUnit = if (StringUtils.isNumeric(driverMemory)) {
diff --git a/linkis-engineconn-plugins/spark/src/main/scala/org/apache/linkis/engineplugin/spark/factory/SparkOnceExecutorFactory.scala b/linkis-engineconn-plugins/spark/src/main/scala/org/apache/linkis/engineplugin/spark/factory/SparkOnceExecutorFactory.scala
index 25e2649441..5802a1c859 100644
--- a/linkis-engineconn-plugins/spark/src/main/scala/org/apache/linkis/engineplugin/spark/factory/SparkOnceExecutorFactory.scala
+++ b/linkis-engineconn-plugins/spark/src/main/scala/org/apache/linkis/engineplugin/spark/factory/SparkOnceExecutorFactory.scala
@@ -22,10 +22,16 @@ import org.apache.linkis.engineconn.common.engineconn.EngineConn
 import org.apache.linkis.engineconn.once.executor.OnceExecutor
 import org.apache.linkis.engineconn.once.executor.creation.OnceExecutorFactory
 import org.apache.linkis.engineplugin.spark.context.SparkEngineConnContext
-import org.apache.linkis.engineplugin.spark.executor.SparkSubmitOnceExecutor
+import org.apache.linkis.engineplugin.spark.executor.{
+  SparkOnKubernetesSubmitOnceExecutor,
+  SparkSubmitOnceExecutor
+}
+import org.apache.linkis.manager.common.conf.RMConfiguration.DEFAULT_KUBERNETES_TYPE
 import org.apache.linkis.manager.label.entity.Label
+import org.apache.linkis.manager.label.entity.cluster.ClusterLabel
 import org.apache.linkis.manager.label.entity.engine.RunType
 import org.apache.linkis.manager.label.entity.engine.RunType.RunType
+import org.apache.linkis.manager.label.utils.LabelUtil
 
 class SparkOnceExecutorFactory extends OnceExecutorFactory {
 
@@ -34,11 +40,24 @@ class SparkOnceExecutorFactory extends OnceExecutorFactory {
       engineCreationContext: EngineCreationContext,
       engineConn: EngineConn,
       labels: Array[Label[_]]
-  ): OnceExecutor =
+  ): OnceExecutor = {
+    val clusterLabel = LabelUtil.getLabelFromArray[ClusterLabel](labels)
     engineConn.getEngineConnSession match {
       case context: SparkEngineConnContext =>
-        new SparkSubmitOnceExecutor(id, context)
+        if (
+            null != clusterLabel && clusterLabel.getClusterType.equalsIgnoreCase(
+              DEFAULT_KUBERNETES_TYPE.getValue
+            )
+        ) {
+          new SparkOnKubernetesSubmitOnceExecutor(id, context)
+        } else {
+          new SparkSubmitOnceExecutor(id, context)
+        }
     }
+  }
+
+  override protected def getSupportRunTypes: Array[String] =
+    Array(RunType.JAR.toString, RunType.PYSPARK.toString)
 
   override protected def getRunType: RunType = RunType.JAR
 }
diff --git a/linkis-engineconn-plugins/spark/src/main/scala/org/apache/linkis/engineplugin/spark/utils/SparkJobProgressUtil.scala b/linkis-engineconn-plugins/spark/src/main/scala/org/apache/linkis/engineplugin/spark/utils/SparkJobProgressUtil.scala
index 196414420a..94614f902b 100644
--- a/linkis-engineconn-plugins/spark/src/main/scala/org/apache/linkis/engineplugin/spark/utils/SparkJobProgressUtil.scala
+++ b/linkis-engineconn-plugins/spark/src/main/scala/org/apache/linkis/engineplugin/spark/utils/SparkJobProgressUtil.scala
@@ -27,11 +27,16 @@ import org.apache.http.client.methods.HttpGet
 import org.apache.http.impl.client.HttpClients
 import org.apache.http.util.EntityUtils
 
+import java.util
+
 object SparkJobProgressUtil extends Logging {
 
-  def getProgress(applicationId: String): Float = {
+  def getProgress(applicationId: String, sparkUIUrl: String = ""): Float = {
     if (StringUtils.isBlank(applicationId)) return 0f
-    val sparkJobsResult = getSparkJobInfo(applicationId)
+    val sparkJobsResult =
+      if (StringUtils.isBlank(sparkUIUrl))
+        getSparkJobInfo(applicationId)
+      else getKubernetesSparkJobInfo(applicationId, sparkUIUrl)
     if (sparkJobsResult.isEmpty) return 0f
     val tuple = sparkJobsResult
       .filter(sparkJobResult => {
@@ -48,8 +53,14 @@ object SparkJobProgressUtil extends Logging {
     tuple._2.toFloat / tuple._1
   }
 
-  def getSparkJobProgressInfo(applicationId: String): Array[JobProgressInfo] = {
-    val sparkJobsResult = getSparkJobInfo(applicationId)
+  def getSparkJobProgressInfo(
+      applicationId: String,
+      sparkUIUrl: String = ""
+  ): Array[JobProgressInfo] = {
+    val sparkJobsResult =
+      if (StringUtils.isBlank(sparkUIUrl))
+        getSparkJobInfo(applicationId)
+      else getKubernetesSparkJobInfo(applicationId, sparkUIUrl)
     if (sparkJobsResult.isEmpty) {
       Array.empty
     } else {
@@ -96,6 +107,38 @@ object SparkJobProgressUtil extends Logging {
       )
     }
 
+  def getKubernetesSparkJobInfo(
+      applicationId: String,
+      sparkUIUrl: String
+  ): Array[java.util.Map[String, Object]] =
+    if (StringUtils.isBlank(applicationId) || StringUtils.isBlank(sparkUIUrl)) Array.empty
+    else {
+      val getSparkJobsStateUrl = s"http://$sparkUIUrl/api/v1/applications/$applicationId"
+      logger.info(s"get spark job state from kubernetes spark ui, url: $getSparkJobsStateUrl")
+      val appStateResult =
+        JsonUtils.jackson.readValue(
+          get(getSparkJobsStateUrl),
+          classOf[java.util.Map[String, Object]]
+        )
+      val appAttemptList = appStateResult.get("attempts").asInstanceOf[java.util.List[Object]]
+      if (appAttemptList == null || appAttemptList.size() == 0) return Array.empty
+      val appLastAttempt =
+        appAttemptList.get(appAttemptList.size() - 1).asInstanceOf[util.Map[String, Object]]
+      val isLastAttemptCompleted = appLastAttempt.get("completed").asInstanceOf[Boolean]
+      if (isLastAttemptCompleted) return Array.empty
+      val getSparkJobsInfoUrl =
+        s"http://$sparkUIUrl/api/v1/applications/$applicationId/jobs"
+      logger.info(s"get spark job info from kubernetes spark ui: $getSparkJobsInfoUrl")
+      val jobs = get(getSparkJobsInfoUrl)
+      if (StringUtils.isBlank(jobs)) {
+        return Array.empty
+      }
+      JsonUtils.jackson.readValue(
+        get(getSparkJobsInfoUrl),
+        classOf[Array[java.util.Map[String, Object]]]
+      )
+    }
+
   def get(url: String): String = {
     val httpGet = new HttpGet(url)
     val client = HttpClients.createDefault
diff --git a/linkis-public-enhancements/linkis-pes-client/src/main/scala/org/apache/linkis/bml/client/impl/HttpBmlClient.scala b/linkis-public-enhancements/linkis-pes-client/src/main/scala/org/apache/linkis/bml/client/impl/HttpBmlClient.scala
index be251d3b23..258b3feb9b 100644
--- a/linkis-public-enhancements/linkis-pes-client/src/main/scala/org/apache/linkis/bml/client/impl/HttpBmlClient.scala
+++ b/linkis-public-enhancements/linkis-pes-client/src/main/scala/org/apache/linkis/bml/client/impl/HttpBmlClient.scala
@@ -137,15 +137,6 @@ class HttpBmlClient(
     if (version != null) bmlDownloadAction.getParameters.asScala += "version" -> version
     bmlDownloadAction.setUser(user)
     val downloadResult = dwsClient.execute(bmlDownloadAction)
-    //    val retIs = new ByteArrayInputStream(IOUtils.toString(bmlDownloadAction.getInputStream).getBytes("UTF-8"))
-    //    if (downloadResult != null) {
-    //      bmlDownloadAction.getResponse match {
-    //        case r: CloseableHttpResponse =>
-    //          Utils.tryAndWarn(r.close())
-    //        case o: Any =>
-    //          info(s"Download response : ${o.getClass.getName} cannot close.")
-    //      }
-    //    }
     BmlDownloadResponse(
       isSuccess = true,
       bmlDownloadAction.getInputStream,
diff --git a/pom.xml b/pom.xml
index 9ef6835630..e4d07644cf 100644
--- a/pom.xml
+++ b/pom.xml
@@ -131,6 +131,7 @@
     <openlookeng.version>1.5.0</openlookeng.version>
     <pipeline.version>1</pipeline.version>
     <presto.version>0.234</presto.version>
+    <nebula.version>3.0.0</nebula.version>
     <python.version>python2</python.version>
     <seatunnel.version>2.1.2</seatunnel.version>
     <shell.version>1</shell.version>