使用代码管理 spark作业: submit, kill

系统管理员 2022-04-22 15:38 145阅读 0赞

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V5ZW9mZWFnbGU_size_16_color_FFFFFF_t_70]

# 1, spark on yarn #

## a， job提交 ##

import org.apache.spark.launcher.SparkLauncher;
    
    import java.io.IOException;
    import java.util.HashMap;
    
    public class Submit_launcher {
    
        public static void main(String[] args) throws IOException, InterruptedException {
            if (args.length < 7) {
                System.err.println("至少7个参数： sparkHome,jobName, mainClassName," +
                        "\r\n cluster, deploy_mode  " +
                        "\r\n jarPath, hbase表名");
                System.err.println("第8 个参数可选： hbase表的字段名");
                System.exit(1);
            }
            System.setProperty("user.name", "hdfs");//设置hdfs用户提交spark任务
            HashMap env = new HashMap();
            //这两个属性必须设置
            env.put("HADOOP_CONF_DIR", "/etc/hadoop/conf");
            env.put("YARN_CONF_DIR", "/etc/hadoop/conf.cloudera.yarn");
            env.put("SPARK_CONF_DIR", "/etc/spark/conf");
    
            String sparkHome = args[0];//"/opt/cloudera/parcels/CDH-5.12.0-1.cdh5.12.0.p0.29/lib/spark"
            String jobName = args[1];
            String mainClassName = args[2];
            String cluster = args[3];
            String deploy_mode = args[4];
    
            String jarPath = args[5];
            String tableName = args[6];
            String propName = "";
            if (args.length > 7) {
                propName = args[7];
            }
            //---------提交------------
            SparkLauncher handle = new SparkLauncher(env)
                    .setSparkHome(sparkHome)
                    .setAppName(jobName)
                    .setMainClass(mainClassName).addAppArgs(tableName, propName)
                    .setMaster(cluster)
                    .setAppResource(jarPath)
    //                .setConf("spark.app.id", "11222")
    //                .setConf("spark.driver.memory", "2g") //SparkLauncher.DRIVER_MEMORY
    //                .setConf("spark.akka.frameSize", "200")
    //                .setConf("spark.executor.memory", "1g")
    //                .setConf("spark.executor.instances", "32")
    //                .setConf("spark.executor.cores", "3")
    //                .setConf("spark.default.parallelism", "10")
    //                .setConf("spark.driver.allowMultipleContexts", "true")
                    .setVerbose(true);
            if (deploy_mode != null && deploy_mode.trim() != "") {
                handle.setDeployMode(deploy_mode);
            }
    
            //---------获取提交结果----------
            Process process = handle.launch();
            InputStreamReaderRunnable inputStreamReaderRunnable = new InputStreamReaderRunnable(process.getInputStream(), "input");
            Thread inputThread = new Thread(inputStreamReaderRunnable, "LogStreamReader input");
            inputThread.start();
    
            InputStreamReaderRunnable errorStreamReaderRunnable = new InputStreamReaderRunnable(process.getErrorStream(), "error");
            Thread errorThread = new Thread(errorStreamReaderRunnable, "LogStreamReader error");
            errorThread.start();
    
            System.out.println("Waiting for finish...");
            int exitCode = process.waitFor();
            System.out.println("Finished! Exit code:" + exitCode);
        }
    }

线程类： 打印job的工作日志

import java.io.BufferedReader;
    import java.io.IOException;
    import java.io.InputStream;
    import java.io.InputStreamReader;
    
    public class InputStreamReaderRunnable implements Runnable {
        private BufferedReader reader;
        private String name;
    
        public InputStreamReaderRunnable(InputStream is, String name) {
            this.reader = new BufferedReader(new InputStreamReader(is));
            this.name = name;
        }
    
        public void run() {
            System.out.println("InputStream ============线程启动" + name + ":");
            try {
                String line = reader.readLine();
                while (line != null) {
                    System.out.println(line);
                    line = reader.readLine();
                }
                reader.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

## b， job的停止 ##

import org.apache.hadoop.conf.Configuration;
    import org.apache.hadoop.yarn.api.records.ApplicationId;
    import org.apache.hadoop.yarn.api.records.ApplicationReport;
    import org.apache.hadoop.yarn.api.records.YarnApplicationState;
    import org.apache.hadoop.yarn.client.api.YarnClient;
    import org.apache.hadoop.yarn.client.cli.ApplicationCLI;
    import org.apache.hadoop.yarn.exceptions.ApplicationNotFoundException;
    import org.apache.hadoop.yarn.exceptions.YarnException;
    import org.apache.hadoop.yarn.util.ConverterUtils;
    
    import java.io.IOException;
    import java.util.EnumSet;
    import java.util.HashSet;
    import java.util.List;
    import java.util.Set;
    
    public class JobKill {
    
        public static void main(String[] args) throws Exception {
            String jobName="";
            if (args.length >0){
                jobName=args[0];
                getJobIdAndKill(jobName);
            }else {
                System.err.println("请输入job名称, 来杀死job.....");
            }
        }
    
        private static void getJobIdAndKill(String jobName) throws IOException, YarnException {
            ApplicationCLI cli = new ApplicationCLI();
            cli.setSysOutPrintStream(System.out);
            cli.setSysErrPrintStream(System.err);
    
            Set<String> appTypes = new HashSet<String>();
            EnumSet<YarnApplicationState> appStates = EnumSet
                    .noneOf(YarnApplicationState.class);
            appStates.add(YarnApplicationState.RUNNING);
            appStates.add(YarnApplicationState.ACCEPTED);
            appStates.add(YarnApplicationState.SUBMITTED);
    
            //yarn客户端： 获取job信息
            YarnClient  client = YarnClient.createYarnClient();
            client.init(new Configuration());
            client.start();
            List<ApplicationReport> appsReport = client.getApplications(appTypes, appStates);
            for (ApplicationReport rep: appsReport){
                String appName = rep.getName();
                ApplicationId appId = rep.getApplicationId();
                System.out.println("jobName=> "+appName+",  jobId=> "+appId.toString());
                if (jobName.equalsIgnoreCase(appName)){
                    killApplication(appId.toString(), client);
                }
            }
        }
    
        public static void killApplication(String applicationId, YarnClient client) throws YarnException,
                IOException {
            ApplicationId appId = ConverterUtils.toApplicationId(applicationId);
            ApplicationReport  appReport = null;
            try {
                appReport = client.getApplicationReport(appId);
            } catch (ApplicationNotFoundException e) {
                System.out.println("Application with id '" + applicationId + "' doesn't exist in RM.");
                throw e;
            }
            if (appReport.getYarnApplicationState() == YarnApplicationState.FINISHED
                    || appReport.getYarnApplicationState() == YarnApplicationState.KILLED
                    || appReport.getYarnApplicationState() == YarnApplicationState.FAILED) {
                System.out.println("Application " + applicationId + " has already finished ");
            } else {
                System.out.println("Killing application " + applicationId);
                client.killApplication(appId);
            }
        }
    }

# 2, spark local 模式： 停止job #

object Test {
      def main(args: Array[String]): Unit = {
        //spark配置
        val conf = new SparkConf()
        conf.setAppName("count")
        conf.setMaster("local[*]")
    
        val sc = new SparkContext(conf)
        sc.addSparkListener(new MyListener())
        sc.setLocalProperty("thread", "client")
        sc.setLocalProperty("spark.job.interruptOnCancel", "true")
    
        val cnt = sc.makeRDD(1 to 300).map(x => {Thread.sleep(1000);x}).count()
        println("count ----> "+cnt)
      }
    }

通过钩子线程，获取Job提交时的jobid :

import org.apache.log4j.Logger;
    import org.apache.spark.scheduler.*;
    import java.util.concurrent.ConcurrentHashMap;
    
    public class MyListener implements SparkListener {
        private static Logger logger = Logger.getLogger(MyListener.class);
        // 存储了提交job的线程局部变量和job的映射关系
        private static ConcurrentHashMap<String, Integer> jobInfoMap= new ConcurrentHashMap<String, Integer>();
    
        public void onJobEnd(SparkListenerJobEnd jobEnd) {
            System.out.println(" Job End:" + jobEnd.jobResult().getClass() + ",Id:" + jobEnd.jobId());
            for (String key : jobInfoMap.keySet()) {
                if (jobInfoMap.get(key) == jobEnd.jobId()) {
                    jobInfoMap.remove(key);
                    System.out.println(key + " request has been returned. because " + jobEnd.jobResult().getClass());
                }
            }
        }
    
        public void onJobStart(SparkListenerJobStart jobStart) {
            System.out.println(" Job Start: JobId->"  + jobStart.jobId());
            //根据线程变量属性找到该job是哪个线程提交的
            System.out.println("MyListener Job Start: Thread->" + jobStart.properties().getProperty("thread", "default"));
            jobInfoMap.put(jobStart.properties().getProperty("thread", "default"), jobStart.jobId());
        }
    ......
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V5ZW9mZWFnbGU_size_16_color_FFFFFF_t_70 1]

Job Start: JobId->0
    MyListener Job Start: Thread->client
    
    // curl -X POST "http://localhost:4040/stages/stage/kill/?id=0&terminate=true"
    .......
    
    Job End:class org.apache.spark.scheduler.JobFailed,Id:0
    client request has been returned. because class org.apache.spark.scheduler.JobFailed

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V5ZW9mZWFnbGU_size_16_color_FFFFFF_t_70]: /images/20220327/792a403a36d2403db14da177eff04574.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2V5ZW9mZWFnbGU_size_16_color_FFFFFF_t_70 1]: /images/20220327/16e01cedccfa434baa50ac9131a57a70.png