[Feature-14251][Task] Support yarn queue definition in yarn task (#14310)

docs/docs/en/guide/task/flink.md

@@ -35,6 +35,7 @@ Flink task type, used to execute Flink programs. For Flink nodes:
 | TaskManager memory size | Used to set the size of taskManager memories, which can be set according to the actual production environment.                                                                                                                                                          |
 | Number of TaskManager   | Used to set the number of taskManagers, which can be set according to the actual production environment.                                                                                                                                                                |
 | Parallelism             | Used to set the degree of parallelism for executing Flink tasks.                                                                                                                                                                                                        |
+| Yarn queue              | Used to set the yarn queue, use `default` queue by default.                                                                                                                                                                                                             |
 | Main program parameters | Set the input parameters for the Flink program and support the substitution of custom parameter variables.                                                                                                                                                              |
 | Optional parameters     | Support `--jar`, `--files`,` --archives`, `--conf` format.                                                                                                                                                                                                              |
 | Custom parameter        | It is a local user-defined parameter for Flink, and will replace the content with `${variable}` in the script.                                                                                                                                                          |

docs/docs/en/guide/task/map-reduce.md

@@ -30,6 +30,7 @@ MapReduce(MR) task type used for executing MapReduce programs. For MapReduce nod
 | The class of the main function | The **full path** of Main Class, the entry point of the MapReduce program.                                         |
 | Main jar package               | The jar package of the MapReduce program.                                                                          |
 | Task name                      | MapReduce task name.                                                                                               |
+| Yarn queue                     | Used to set the yarn queue, use `default` queue by default.                                                        |
 | Command line parameters        | Set the input parameters of the MapReduce program and support the substitution of custom parameter variables.      |
 | Other parameters               | Support `-D`, `-files`, `-libjars`, `-archives` format.                                                            |
 | User-defined parameter         | It is a local user-defined parameter for MapReduce, and will replace the content with `${variable}` in the script. |

docs/docs/en/guide/task/spark.md

@@ -33,6 +33,7 @@ Spark task type for executing Spark application. When executing the Spark task,
 | Driver memory size         | Set the size of Driver memories, which can be set according to the actual production environment.                                  |
 | Number of Executor         | Set the number of Executor, which can be set according to the actual production environment.                                       |
 | Executor memory size       | Set the size of Executor memories, which can be set according to the actual production environment.                                |
+| Yarn queue                 | Set the yarn queue, use `default` queue by default.                                                                                |
 | Main program parameters    | Set the input parameters of the Spark program and support the substitution of custom parameter variables.                          |
 | Optional parameters        | Support `--jars`, `--files`,` --archives`, `--conf` format.                                                                        |
 | Resource                   | Appoint resource files in the `Resource` if parameters refer to them.                                                              |

docs/docs/zh/guide/task/flink.md

@@ -35,6 +35,7 @@ Flink 任务类型，用于执行 Flink 程序。对于 Flink 节点：
 | taskManager 内存数 | 用于设置 taskManager 内存数，可根据实际生产环境设置对应的内存数                                                                                                                                                                                                     |
 | taskManager 数量  | 用于设置 taskManager 的数量，可根据实际生产环境设置对应的数量                                                                                                                                                                                                      |
 | 并行度             | 用于设置执行 Flink 任务的并行度                                                                                                                                                                                                                        |
+| Yarn 队列         | 用于设置 Yarn 队列，默认使用 default 队列                                                                                                                                                                                                               |
 | 主程序参数           | 设置 Flink 程序的输入参数，支持自定义参数变量的替换                                                                                                                                                                                                              |
 | 选项参数            | 支持 `--jar`、`--files`、`--archives`、`--conf` 格式                                                                                                                                                                                              |
 | 自定义参数           | 是 Flink 局部的用户自定义参数，会替换脚本中以 ${变量} 的内容                                                                                                                                                                                                       |

docs/docs/zh/guide/task/map-reduce.md

@@ -24,6 +24,7 @@ MapReduce(MR) 任务类型，用于执行 MapReduce 程序。对于 MapReduce
 | 主函数的 Class | 是 MapReduce 程序的入口 Main Class 的**全路径**    |
 | 主程序包       | 执行 MapReduce 程序的 jar 包                   |
 | 任务名称（选填）   | MapReduce 任务名称                           |
+| Yarn 队列    | 设置 Yarn 队列，默认使用 default                  |
 | 命令行参数      | 是设置 MapReduce 程序的输入参数，支持自定义参数变量的替换       |
 | 其他参数       | 支持 –D、-files、-libjars、-archives 格式       |
 | 自定义参数      | 是 MapReduce 局部的用户自定义参数，会替换脚本中以 ${变量} 的内容 |

docs/docs/zh/guide/task/spark.md

@@ -32,6 +32,7 @@ Spark  任务类型用于执行 Spark 应用。对于 Spark 节点，worker 支
 - Driver 内存数：用于设置 Driver 内存数，可根据实际生产环境设置对应的内存数。
 - Executor 数量：用于设置 Executor 的数量，可根据实际生产环境设置对应的内存数。
 - Executor 内存数：用于设置 Executor 内存数，可根据实际生产环境设置对应的内存数。
+- Yarn 队列：用于设置 Yarn 队列，默认使用 default 队列。
 - 主程序参数：设置 Spark 程序的输入参数，支持自定义参数变量的替换。
 - 选项参数：支持 `--jars`、`--files`、`--archives`、`--conf` 格式。
 - 资源：如果其他参数中引用了资源文件，需要在资源中选择指定。

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/builder/TaskExecutionContextBuilder.java

@@ -106,7 +106,6 @@ public class TaskExecutionContextBuilder {
         taskExecutionContext.setExecutorId(processInstance.getExecutorId());
         taskExecutionContext.setCmdTypeIfComplement(processInstance.getCmdTypeIfComplement().getCode());
         taskExecutionContext.setTenantCode(processInstance.getTenantCode());
-        taskExecutionContext.setQueue(processInstance.getQueue());
         return this;
     }
 

dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/TaskExecutionContext.java

@@ -146,11 +146,6 @@ public class TaskExecutionContext implements Serializable {
      */
     private String tenantCode;
 
-    /**
-     * task queue
-     */
-    private String queue;
-
     /**
      * process define id
      */

dolphinscheduler-task-plugin/dolphinscheduler-task-api/src/main/java/org/apache/dolphinscheduler/plugin/task/api/parameters/dataquality/spark/SparkParameters.java

@@ -82,7 +82,7 @@ public class SparkParameters extends AbstractParameters {
     /**
      * The YARN queue to submit to
      */
-    private String queue;
+    private String yarnQueue;
 
     /**
      * other arguments
@@ -180,12 +180,12 @@ public class SparkParameters extends AbstractParameters {
         this.appName = appName;
     }
 
-    public String getQueue() {
-        return queue;
+    public String getYarnQueue() {
+        return yarnQueue;
     }
 
-    public void setQueue(String queue) {
-        this.queue = queue;
+    public void setYarnQueue(String yarnQueue) {
+        this.yarnQueue = yarnQueue;
     }
 
     public String getOthers() {

dolphinscheduler-task-plugin/dolphinscheduler-task-dataquality/src/main/java/org/apache/dolphinscheduler/plugin/task/dq/DataQualityTask.java

@@ -121,10 +121,6 @@ public class DataQualityTask extends AbstractYarnTask {
                                 StringEscapeUtils.escapeJava(JSONUtils.toJsonString(dataQualityConfiguration)))
                         + "\"");
 
-        dataQualityParameters
-                .getSparkParameters()
-                .setQueue(dqTaskExecutionContext.getQueue());
-
         setMainJarName();
     }
 

dolphinscheduler-task-plugin/dolphinscheduler-task-dataquality/src/main/java/org/apache/dolphinscheduler/plugin/task/dq/utils/SparkArgsUtils.java

@@ -106,7 +106,7 @@ public class SparkArgsUtils {
         String others = param.getOthers();
         if (!SPARK_LOCAL.equals(deployMode)
                 && (StringUtils.isEmpty(others) || !others.contains(SparkConstants.SPARK_QUEUE))) {
-            String queue = param.getQueue();
+            String queue = param.getYarnQueue();
             if (StringUtils.isNotEmpty(queue)) {
                 args.add(SparkConstants.SPARK_QUEUE);
                 args.add(queue);

dolphinscheduler-task-plugin/dolphinscheduler-task-flink-stream/src/main/java/org/apache/dolphinscheduler/plugin/task/flink/FlinkStreamTask.java

@@ -56,7 +56,6 @@ public class FlinkStreamTask extends FlinkTask implements StreamTask {
         if (flinkParameters == null || !flinkParameters.checkParameters()) {
             throw new RuntimeException("flink task params is not valid");
         }
-        flinkParameters.setQueue(taskExecutionContext.getQueue());
 
         FileUtils.generateScriptFile(taskExecutionContext, flinkParameters);
     }

dolphinscheduler-task-plugin/dolphinscheduler-task-flink/src/main/java/org/apache/dolphinscheduler/plugin/task/flink/FlinkArgsUtils.java

@@ -163,9 +163,9 @@ public class FlinkArgsUtils {
             }
 
             // yarn.application.queue
-            String queue = flinkParameters.getQueue();
-            if (StringUtils.isNotEmpty(queue)) {
-                initOptions.add(String.format(FlinkConstants.FLINK_FORMAT_YARN_APPLICATION_QUEUE, queue));
+            String yarnQueue = flinkParameters.getYarnQueue();
+            if (StringUtils.isNotEmpty(yarnQueue)) {
+                initOptions.add(String.format(FlinkConstants.FLINK_FORMAT_YARN_APPLICATION_QUEUE, yarnQueue));
             }
         }
 
@@ -306,26 +306,26 @@ public class FlinkArgsUtils {
             case CLUSTER:
                 if (FLINK_VERSION_AFTER_OR_EQUALS_1_12.equals(flinkVersion)
                         || FLINK_VERSION_AFTER_OR_EQUALS_1_13.equals(flinkVersion)) {
-                    doAddQueue(args, flinkParameters, FlinkConstants.FLINK_QUEUE_FOR_TARGETS);
+                    doAddQueue(args, flinkParameters, FlinkConstants.FLINK_YARN_QUEUE_FOR_TARGETS);
                 } else {
-                    doAddQueue(args, flinkParameters, FlinkConstants.FLINK_QUEUE_FOR_MODE);
+                    doAddQueue(args, flinkParameters, FlinkConstants.FLINK_YARN_QUEUE_FOR_MODE);
                 }
             case APPLICATION:
-                doAddQueue(args, flinkParameters, FlinkConstants.FLINK_QUEUE_FOR_TARGETS);
+                doAddQueue(args, flinkParameters, FlinkConstants.FLINK_YARN_QUEUE_FOR_TARGETS);
         }
     }
 
     private static void doAddQueue(List<String> args, FlinkParameters flinkParameters, String option) {
         String others = flinkParameters.getOthers();
         if (StringUtils.isEmpty(others) || !others.contains(option)) {
-            String queue = flinkParameters.getQueue();
-            if (StringUtils.isNotEmpty(queue)) {
+            String yarnQueue = flinkParameters.getYarnQueue();
+            if (StringUtils.isNotEmpty(yarnQueue)) {
                 switch (option) {
-                    case FlinkConstants.FLINK_QUEUE_FOR_TARGETS:
-                        args.add(String.format(FlinkConstants.FLINK_QUEUE_FOR_TARGETS + "=%s", queue));
-                    case FlinkConstants.FLINK_QUEUE_FOR_MODE:
-                        args.add(FlinkConstants.FLINK_QUEUE_FOR_MODE);
-                        args.add(queue);
+                    case FlinkConstants.FLINK_YARN_QUEUE_FOR_TARGETS:
+                        args.add(String.format(FlinkConstants.FLINK_YARN_QUEUE_FOR_TARGETS + "=%s", yarnQueue));
+                    case FlinkConstants.FLINK_YARN_QUEUE_FOR_MODE:
+                        args.add(FlinkConstants.FLINK_YARN_QUEUE_FOR_MODE);
+                        args.add(yarnQueue);
                 }
             }
         }

dolphinscheduler-task-plugin/dolphinscheduler-task-flink/src/main/java/org/apache/dolphinscheduler/plugin/task/flink/FlinkConstants.java

@@ -48,8 +48,8 @@ public class FlinkConstants {
     public static final String FLINK_EXECUTION_TARGET = "-t";
     public static final String FLINK_YARN_SLOT = "-ys";
     public static final String FLINK_APP_NAME = "-ynm";
-    public static final String FLINK_QUEUE_FOR_MODE = "-yqu";
-    public static final String FLINK_QUEUE_FOR_TARGETS = "-Dyarn.application.queue";
+    public static final String FLINK_YARN_QUEUE_FOR_MODE = "-yqu";
+    public static final String FLINK_YARN_QUEUE_FOR_TARGETS = "-Dyarn.application.queue";
     public static final String FLINK_TASK_MANAGE = "-yn";
     public static final String FLINK_JOB_MANAGE_MEM = "-yjm";
     public static final String FLINK_TASK_MANAGE_MEM = "-ytm";

dolphinscheduler-task-plugin/dolphinscheduler-task-flink/src/main/java/org/apache/dolphinscheduler/plugin/task/flink/FlinkParameters.java

@@ -86,7 +86,7 @@ public class FlinkParameters extends AbstractParameters {
     /**
      * The YARN queue to submit to
      */
-    private String queue;
+    private String yarnQueue;
 
     /**
      * other arguments
@@ -194,12 +194,12 @@ public class FlinkParameters extends AbstractParameters {
         this.taskManagerMemory = taskManagerMemory;
     }
 
-    public String getQueue() {
-        return queue;
+    public String getYarnQueue() {
+        return yarnQueue;
     }
 
-    public void setQueue(String queue) {
-        this.queue = queue;
+    public void setYarnQueue(String yarnQueue) {
+        this.yarnQueue = yarnQueue;
     }
 
     public List<ResourceInfo> getResourceList() {

dolphinscheduler-task-plugin/dolphinscheduler-task-flink/src/main/java/org/apache/dolphinscheduler/plugin/task/flink/FlinkTask.java

@@ -59,7 +59,6 @@ public class FlinkTask extends AbstractYarnTask {
         if (flinkParameters == null || !flinkParameters.checkParameters()) {
             throw new RuntimeException("flink task params is not valid");
         }
-        flinkParameters.setQueue(taskExecutionContext.getQueue());
 
         FileUtils.generateScriptFile(taskExecutionContext, flinkParameters);
     }

dolphinscheduler-task-plugin/dolphinscheduler-task-mr/src/main/java/org/apache/dolphinscheduler/plugin/task/mr/MapReduceArgsUtils.java

@@ -20,7 +20,7 @@ package org.apache.dolphinscheduler.plugin.task.mr;
 import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.D;
 import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.JAR;
 import static org.apache.dolphinscheduler.plugin.task.mr.MapReduceTaskConstants.MR_NAME;
-import static org.apache.dolphinscheduler.plugin.task.mr.MapReduceTaskConstants.MR_QUEUE;
+import static org.apache.dolphinscheduler.plugin.task.mr.MapReduceTaskConstants.MR_YARN_QUEUE;
 
 import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
 import org.apache.dolphinscheduler.plugin.task.api.model.ResourceInfo;
@@ -67,10 +67,10 @@ public class MapReduceArgsUtils {
         }
 
         String others = param.getOthers();
-        if (StringUtils.isEmpty(others) || !others.contains(MR_QUEUE)) {
-            String queue = param.getQueue();
-            if (StringUtils.isNotEmpty(queue)) {
-                args.add(String.format("%s%s=%s", D, MR_QUEUE, queue));
+        if (StringUtils.isEmpty(others) || !others.contains(MR_YARN_QUEUE)) {
+            String yarnQueue = param.getYarnQueue();
+            if (StringUtils.isNotEmpty(yarnQueue)) {
+                args.add(String.format("%s%s=%s", D, MR_YARN_QUEUE, yarnQueue));
             }
         }
 

dolphinscheduler-task-plugin/dolphinscheduler-task-mr/src/main/java/org/apache/dolphinscheduler/plugin/task/mr/MapReduceParameters.java

@@ -54,9 +54,9 @@ public class MapReduceParameters extends AbstractParameters {
     private String appName;
 
     /**
-     * queue
+     * The YARN queue to submit to
      */
-    private String queue;
+    private String yarnQueue;
 
     /**
      * resource list
@@ -101,12 +101,12 @@ public class MapReduceParameters extends AbstractParameters {
         this.appName = appName;
     }
 
-    public String getQueue() {
-        return queue;
+    public String getYarnQueue() {
+        return yarnQueue;
     }
 
-    public void setQueue(String queue) {
-        this.queue = queue;
+    public void setYarnQueue(String yarnQueue) {
+        this.yarnQueue = yarnQueue;
     }
 
     public List<ResourceInfo> getResourceList() {
@@ -152,7 +152,7 @@ public class MapReduceParameters extends AbstractParameters {
         return "mainJar= " + mainJar
                 + "mainClass=" + mainClass
                 + "mainArgs=" + mainArgs
-                + "queue=" + queue
+                + "yarnQueue=" + yarnQueue
                 + "other mainArgs=" + others;
     }
 }

dolphinscheduler-task-plugin/dolphinscheduler-task-mr/src/main/java/org/apache/dolphinscheduler/plugin/task/mr/MapReduceTask.java

@@ -70,8 +70,6 @@ public class MapReduceTask extends AbstractYarnTask {
             throw new RuntimeException("mapreduce task params is not valid");
         }
 
-        mapreduceParameters.setQueue(taskExecutionContext.getQueue());
-
         // replace placeholder,and combine local and global parameters
         Map<String, Property> paramsMap = taskExecutionContext.getPrepareParamsMap();
 

dolphinscheduler-task-plugin/dolphinscheduler-task-mr/src/main/java/org/apache/dolphinscheduler/plugin/task/mr/MapReduceTaskConstants.java

@@ -31,6 +31,6 @@ public class MapReduceTaskConstants {
     /**
      * -D mapreduce.job.queuename=queuename
      */
-    public static final String MR_QUEUE = "mapreduce.job.queuename";
+    public static final String MR_YARN_QUEUE = "mapreduce.job.queuename";
 
 }

dolphinscheduler-task-plugin/dolphinscheduler-task-spark/src/main/java/org/apache/dolphinscheduler/plugin/task/spark/SparkConstants.java

@@ -36,7 +36,7 @@ public class SparkConstants {
     /**
      * --queue QUEUE
      */
-    public static final String SPARK_QUEUE = "--queue";
+    public static final String SPARK_YARN_QUEUE = "--queue";
 
     public static final String DEPLOY_MODE = "--deploy-mode";
 

dolphinscheduler-task-plugin/dolphinscheduler-task-spark/src/main/java/org/apache/dolphinscheduler/plugin/task/spark/SparkParameters.java

@@ -82,7 +82,7 @@ public class SparkParameters extends AbstractParameters {
     /**
      * The YARN queue to submit to
      */
-    private String queue;
+    private String yarnQueue;
 
     /**
      * other arguments
@@ -198,12 +198,12 @@ public class SparkParameters extends AbstractParameters {
         this.appName = appName;
     }
 
-    public String getQueue() {
-        return queue;
+    public String getYarnQueue() {
+        return yarnQueue;
     }
 
-    public void setQueue(String queue) {
-        this.queue = queue;
+    public void setYarnQueue(String yarnQueue) {
+        this.yarnQueue = yarnQueue;
     }
 
     public String getOthers() {

dolphinscheduler-task-plugin/dolphinscheduler-task-spark/src/main/java/org/apache/dolphinscheduler/plugin/task/spark/SparkTask.java

@@ -81,7 +81,6 @@ public class SparkTask extends AbstractYarnTask {
         if (!sparkParameters.checkParameters()) {
             throw new RuntimeException("spark task params is not valid");
         }
-        sparkParameters.setQueue(taskExecutionContext.getQueue());
 
         log.info("Initialize spark task params {}", JSONUtils.toPrettyJsonString(sparkParameters));
     }
@@ -165,11 +164,11 @@ public class SparkTask extends AbstractYarnTask {
 
         String others = sparkParameters.getOthers();
         if (!SparkConstants.DEPLOY_MODE_LOCAL.equals(deployMode)
-                && (StringUtils.isEmpty(others) || !others.contains(SparkConstants.SPARK_QUEUE))) {
-            String queue = sparkParameters.getQueue();
-            if (StringUtils.isNotEmpty(queue)) {
-                args.add(SparkConstants.SPARK_QUEUE);
-                args.add(queue);
+                && (StringUtils.isEmpty(others) || !others.contains(SparkConstants.SPARK_YARN_QUEUE))) {
+            String yarnQueue = sparkParameters.getYarnQueue();
+            if (StringUtils.isNotEmpty(yarnQueue)) {
+                args.add(SparkConstants.SPARK_YARN_QUEUE);
+                args.add(yarnQueue);
             }
         }
 

dolphinscheduler-ui/src/locales/en_US/project.ts

@@ -854,6 +854,8 @@ export default {
     dynamic_separator_tips: 'separator(required)',
     child_node_definition: 'child node definition',
     child_node_instance: 'child node instance',
+    yarn_queue: 'Yarn Queue',
+    yarn_queue_tips: 'Please input yarn queue(optional)',
   },
   menu: {
     fav: 'Favorites',

dolphinscheduler-ui/src/locales/zh_CN/project.ts

@@ -830,6 +830,8 @@ export default {
     dynamic_separator_tips: '分隔符(必填)',
     child_node_definition: '子节点定义',
     child_node_instance: '子节点实例',
+    yarn_queue: 'Yarn队列',
+    yarn_queue_tips: '请输入Yarn队列(选填)',
   },
   menu: {
     fav: '收藏组件',

dolphinscheduler-ui/src/views/projects/task/components/node/fields/index.ts

@@ -90,3 +90,4 @@ export { useLinkis } from './use-linkis'
 export { useDataFactory } from './use-data-factory'
 export { useRemoteShell } from './use-remote-shell'
 export { useDynamic } from './use-dynamic'
+export { useYarnQueue } from './use-queue'

dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-flink.ts

@@ -16,7 +16,7 @@
  */
 import { computed, watch, watchEffect } from 'vue'
 import { useI18n } from 'vue-i18n'
-import { useCustomParams, useMainJar, useResources } from '.'
+import { useCustomParams, useMainJar, useResources, useYarnQueue } from '.'
 import type { IJsonItem } from '../types'
 
 export function useFlink(model: { [field: string]: any }): IJsonItem[] {
@@ -282,6 +282,7 @@ export function useFlink(model: { [field: string]: any }): IJsonItem[] {
       },
       value: model.parallelism
     },
+    useYarnQueue(),
     {
       type: 'input',
       field: 'mainArgs',

dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-mr.ts

@@ -16,7 +16,7 @@
  */
 import { computed } from 'vue'
 import { useI18n } from 'vue-i18n'
-import { useCustomParams, useMainJar, useResources } from '.'
+import { useCustomParams, useMainJar, useResources, useYarnQueue } from '.'
 import type { IJsonItem } from '../types'
 
 export function useMr(model: { [field: string]: any }): IJsonItem[] {
@@ -68,6 +68,7 @@ export function useMr(model: { [field: string]: any }): IJsonItem[] {
         placeholder: t('project.node.app_name_tips')
       }
     },
+    useYarnQueue(),
     {
       type: 'input',
       field: 'mainArgs',

dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-queue.ts

@@ -0,0 +1,32 @@
+/*
+ * 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.
+ */
+import { useI18n } from 'vue-i18n'
+import type { IJsonItem } from '../types'
+
+export function useYarnQueue(): IJsonItem {
+  const { t } = useI18n()
+
+  return {
+    type: 'input',
+    field: 'yarnQueue',
+    name: t('project.node.yarn_queue'),
+    span: 12,
+    props: {
+      placeholder: t('project.node.yarn_queue_tips')
+    },
+  }
+}

dolphinscheduler-ui/src/views/projects/task/components/node/fields/use-spark.ts

@@ -26,7 +26,8 @@ import {
   useExecutorCores,
   useMainJar,
   useNamespace,
-  useResources
+  useResources,
+  useYarnQueue
 } from '.'
 import type { IJsonItem } from '../types'
 
@@ -105,6 +106,7 @@ export function useSpark(model: { [field: string]: any }): IJsonItem[] {
     useExecutorNumber(),
     useExecutorMemory(),
     useExecutorCores(),
+    useYarnQueue(),
     {
       type: 'input',
       field: 'mainArgs',

dolphinscheduler-ui/src/views/projects/task/components/node/format-data.ts

@@ -62,6 +62,7 @@ export function formatParams(data: INodeData): {
     if (data.namespace) {
       taskParams.namespace = data.namespace
     }
+    taskParams.yarnQueue = data.yarnQueue
   }
 
   if (data.taskType === 'SPARK') {
@@ -321,7 +322,8 @@ export function formatParams(data: INodeData): {
       executorCores: data.executorCores,
       executorMemory: data.executorMemory,
       numExecutors: data.numExecutors,
-      others: data.others
+      others: data.others,
+      yarnQueue: data.yarnQueue
     }
   }
 

dolphinscheduler-ui/src/views/projects/task/components/node/tasks/use-data-quality.ts

@@ -57,7 +57,8 @@ export function useDataQuality({
     numExecutors: 2,
     executorMemory: '2G',
     executorCores: 2,
-    others: '--conf spark.yarn.maxAppAttempts=1'
+    others: '--conf spark.yarn.maxAppAttempts=1',
+    yarnQueue: ''
   } as INodeData)
 
   return {
@@ -84,6 +85,7 @@ export function useDataQuality({
       Fields.useExecutorNumber(),
       Fields.useExecutorMemory(),
       Fields.useExecutorCores(),
+      Fields.useYarnQueue(),
       {
         type: 'input',
         field: 'others',

dolphinscheduler-ui/src/views/projects/task/components/node/tasks/use-flink-stream.ts

@@ -53,7 +53,8 @@ export function useFlinkStream({
     slot: 1,
     taskManager: 2,
     parallelism: 1,
-    timeoutNotifyStrategy: ['WARN']
+    timeoutNotifyStrategy: ['WARN'],
+    yarnQueue: ''
   })
 
   return {

dolphinscheduler-ui/src/views/projects/task/components/node/tasks/use-flink.ts

@@ -53,7 +53,8 @@ export function useFlink({
     slot: 1,
     taskManager: 2,
     parallelism: 1,
-    timeoutNotifyStrategy: ['WARN']
+    timeoutNotifyStrategy: ['WARN'],
+    yarnQueue: ''
   })
 
   return {

dolphinscheduler-ui/src/views/projects/task/components/node/tasks/use-mr.ts

@@ -44,7 +44,8 @@ export function useMr({
     delayTime: 0,
     timeout: 30,
     programType: 'SCALA',
-    timeoutNotifyStrategy: ['WARN']
+    timeoutNotifyStrategy: ['WARN'],
+    yarnQueue: '',
   } as INodeData)
 
   return {

dolphinscheduler-ui/src/views/projects/task/components/node/tasks/use-spark.ts

@@ -51,6 +51,7 @@ export function useSpark({
     numExecutors: 2,
     executorMemory: '2G',
     executorCores: 2,
+    yarnQueue: '',
     timeoutNotifyStrategy: ['WARN']
   } as INodeData)
 

dolphinscheduler-ui/src/views/projects/task/components/node/types.ts

@@ -223,6 +223,7 @@ interface ISparkParameters {
   executorMemory?: string
   numExecutors?: number
   others?: string
+  yarnQueue?: string
 }
 
 interface IRuleParameters {
@@ -441,6 +442,7 @@ interface ITaskParams {
   degreeOfParallelism?: number
   filterCondition?: string
   listParameters?: Array<any>
+  yarnQueue?: string
 }
 
 interface INodeData