[SparkLoad]Use the yarn command to get status and kill the application

fe/fe-core/src/main/java/org/apache/doris/catalog/SparkResource.java

@@ -19,16 +19,19 @@
 
 import org.apache.doris.analysis.BrokerDesc;
 import org.apache.doris.analysis.ResourceDesc;
+import org.apache.doris.common.Config;
 import org.apache.doris.common.DdlException;
 import org.apache.doris.common.LoadException;
 import org.apache.doris.common.Pair;
 import org.apache.doris.common.proc.BaseProcResult;
 import org.apache.doris.load.loadv2.SparkRepository;
+import org.apache.doris.load.loadv2.SparkYarnConfigFiles;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 import com.google.gson.annotations.SerializedName;
 
+import java.io.File;
 import java.util.Map;
 
 /**
@@ -66,6 +69,7 @@ public class SparkResource extends Resource {
     private static final String SPARK_CONFIG_PREFIX = "spark.";
     private static final String BROKER_PROPERTY_PREFIX = "broker.";
     // spark uses hadoop configs in the form of spark.hadoop.*
+    private static final String SPARK_HADOOP_CONFIG_PREFIX = "spark.hadoop.";
     private static final String SPARK_YARN_RESOURCE_MANAGER_ADDRESS = "spark.hadoop.yarn.resourcemanager.address";
     private static final String SPARK_FS_DEFAULT_FS = "spark.hadoop.fs.defaultFS";
     private static final String YARN_RESOURCE_MANAGER_ADDRESS = "yarn.resourcemanager.address";
@@ -166,6 +170,23 @@ public synchronized SparkRepository.SparkArchive prepareArchive() throws LoadExc
         return archive;
     }
 
+    // Each SparkResource has and only has one yarn config to run yarn command
+    // This method will write all the configuration start with "spark.hadoop." into config files in a specific directory
+    public synchronized String prepareYarnConfig() throws LoadException {
+        SparkYarnConfigFiles yarnConfigFiles = new SparkYarnConfigFiles(name, getSparkHadoopConfig(sparkConfigs));
+        yarnConfigFiles.prepare();
+        return yarnConfigFiles.getConfigDir();
+    }
+
+    public String getYarnClientPath() throws LoadException {
+        String yarnClientPath = Config.yarn_client_path;
+        File file = new File(yarnClientPath);
+        if (!file.exists() || !file.isFile()) {
+            throw new LoadException("yarn client does not exist in path: " + yarnClientPath);
+        }
+        return yarnClientPath;
+    }
+
     public boolean isYarnMaster() {
         return getMaster().equalsIgnoreCase(YARN_MASTER);
     }
@@ -182,7 +203,7 @@ public void update(ResourceDesc resourceDesc) throws DdlException {
         if (properties.containsKey(SPARK_MASTER)) {
             throw new DdlException("Cannot change spark master");
         }
-        sparkConfigs.putAll(getSparkConfigs(properties));
+        sparkConfigs.putAll(getSparkConfig(properties));
 
         // update working dir and broker
         if (properties.containsKey(WORKING_DIR)) {
@@ -199,7 +220,7 @@ protected void setProperties(Map<String, String> properties) throws DdlException
         Preconditions.checkState(properties != null);
 
         // get spark configs
-        sparkConfigs = getSparkConfigs(properties);
+        sparkConfigs = getSparkConfig(properties);
         // check master and deploy mode
         if (getMaster() == null) {
             throw new DdlException("Missing " + SPARK_MASTER + " in properties");
@@ -233,14 +254,24 @@ && isYarnMaster()) {
         brokerProperties = getBrokerProperties(properties);
     }
 
-    private Map<String, String> getSparkConfigs(Map<String, String> properties) {
-        Map<String, String> sparkConfigs = Maps.newHashMap();
+    private Map<String, String> getSparkConfig(Map<String, String> properties) {
+        Map<String, String> sparkConfig = Maps.newHashMap();
         for (Map.Entry<String, String> entry : properties.entrySet()) {
             if (entry.getKey().startsWith(SPARK_CONFIG_PREFIX)) {
-                sparkConfigs.put(entry.getKey(), entry.getValue());
+                sparkConfig.put(entry.getKey(), entry.getValue());
             }
         }
-        return sparkConfigs;
+        return sparkConfig;
+    }
+
+    private Map<String, String> getSparkHadoopConfig(Map<String, String> properties) {
+        Map<String, String> sparkConfig = Maps.newHashMap();
+        for (Map.Entry<String, String> entry : properties.entrySet()) {
+            if (entry.getKey().startsWith(SPARK_HADOOP_CONFIG_PREFIX)) {
+                sparkConfig.put(entry.getKey(), entry.getValue());
+            }
+        }
+        return sparkConfig;
     }
 
     private Map<String, String> getBrokerProperties(Map<String, String> properties) {

fe/fe-core/src/main/java/org/apache/doris/common/Config.java

@@ -543,6 +543,20 @@ public class Config extends ConfigBase {
     @ConfField
     public static String spark_resource_path = "";
 
+    /**
+     * Default yarn client path
+     */
+    @ConfField
+    public static String yarn_client_path = PaloFe.DORIS_HOME_DIR + "/lib/yarn-client/hadoop/bin/yarn";
+
+    /**
+     * Default yarn config file directory
+     * Each time before running the yarn command, we need to check that the
+     * config file exists under this path, and if not, create them.
+     */
+    @ConfField
+    public static String yarn_config_dir = PaloFe.DORIS_HOME_DIR + "/lib/yarn-config";
+
     /**
      * Default number of waiting jobs for routine load and version 2 of load
      * This is a desired number.

fe/fe-core/src/main/java/org/apache/doris/load/loadv2/ConfigFile.java

@@ -0,0 +1,28 @@
+// 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.doris.load.loadv2;
+
+import org.apache.doris.common.LoadException;
+
+// The config file required to run the yarn command.
+// Each time before running the yarn command, we need to check that the
+// config file exists in the specified path, and if not, create them.
+public interface ConfigFile {
+    public void createFile() throws LoadException;
+    public String getFilePath();
+}