SparkLauncher提交spark 正确的退出方式以及状态获取

知其然知其所以然

转载注明出处,且必须看到最后,留言证明

引发问题

spark任务状态获取不准确,任务是失败的,但结果返回成功,在注册的Listener中也可以看到状态先是FINISHED,过一会才会变成FAILED,因为FINISHED是isFinal(),导致任务退出,漏掉了后返回的失败状态

如果采用jobState.isFinal()判断任务结束,则会触发这个问题。

// spark程序退出的错误判断条件
if (jobState.isFinal()) {
 countDownLatch.countDown();
}
                

原理解析

想要从根本上解决问题,还是需要知道为什么会先出现FINISHED后又变成FAILED,或者是怎么变成FAILED。
Spark 任务使用SparkLauncher.startApplication方式提交,核心代码如下,并通过代码逻辑总结为下图关系。
在这里插入图片描述

核心代码:

@Override
public SparkAppHandle startApplication(SparkAppHandle.Listener... listeners) throws IOException {
  // 初始化LauncherServer 
  LauncherServer server = LauncherServer.getOrCreateServer();
  // 初始化AppHandle
  ChildProcAppHandle handle = new ChildProcAppHandle(server);
  // 注册listener
  for (SparkAppHandle.Listener l : listeners) {
    handle.addListener(l);
  }
  // 注册handle
  String secret = server.registerHandle(handle);

  // 初始化进程
  ProcessBuilder pb = createBuilder();

  try {
    // 启动进程
    Process child = pb.start();
    // 注册进程
    handle.setChildProc(child, loggerName, logStream);
  } catch (IOException ioe) {
    handle.kill();
    throw ioe;
  }
  return handle;
}

首先看LauncherServer这个类,进入到类里,查看注释,大致逻辑为当User App(客户端)提交任务到Spark App(服务端),L. Backend会向 L. Server(LauncherServer)响应,之后L. Backend都是与App Handle交互,那么我们就需要重点研究App Handle。

=> LauncherServer类注释
LauncherServer类注释

=> App Handle类分析
从上图总结的关系看出,AppHandle里一共有两个角色,listener和process。listener是我们自定义的,实现stateChanged方法,主要是监听状态变化,那么process是干什么的呢。通过handle.setChildProc()进入到具体实现,通过继续追源码,最终都开启了monitorChild线程

void setChildProc(Process childProc, String loggerName, InputStream logStream) {
  this.childProc = childProc;
  if (logStream != null) {
    this.redirector = new OutputRedirector(logStream, loggerName,
      SparkLauncher.REDIRECTOR_FACTORY, this);
  } else {
    // If there is no log redirection, spawn a thread that will wait for the child process
    // to finish.
    SparkLauncher.REDIRECTOR_FACTORY.newThread(this::monitorChild).start();
  }
}

进入到monitorChild方法,其实就是做一件事,监控SparkLauncher.startApplication里启动的进程process,并且根据进程值做相应状态处理。其中有一句注释

Override state with failure if the current state is not final, or is success

翻译过来就是:

当进程结束时,如果状态是 未完成成功 ,则覆盖(更新)为失败

与问题现象一致,到这,问题根因已经找到,顺便就把问题彻底解决。
从整个逻辑上看,当进程不存在时(proc.isAlive()),并且处理完状态后,最终调用dispose()方法。更新disposed属性为true,在父类AbstractAppHandle中有个isDisposed()方法可以获取该值。

/**
 * Wait for the child process to exit and update the handle's state if necessary, according to
 * the exit code.
 */
void monitorChild() {
  Process proc = childProc;
  // 进程活着则一直等待,进程结束则走下面的处理
  while (proc.isAlive()) {
      proc.waitFor();
  }
  // 获取最终进程值
  int ec;
  try {
      ec = proc.exitValue();
  } catch (Exception e) {
      ec = 1;
  }
  // 根据进程处理状态
  if (ec != 0) {
      State currState = getState();
      // Override state with failure if the current state is not final, or is success.
      if (!currState.isFinal() || currState == State.FINISHED) {
        setState(State.FAILED, true);
      }
  }
  // 释放
  dispose();
  }
}

AbstractAppHandle.isDisposed()方法

boolean isDisposed() {
  return disposed;
}

至此,spark程序是否真正结束应该通过AbstractAppHandle.isDisposed()方法判断
q: 是否可以使用proc.isAlive()方法?
a: 不可以,proc.isAlive()判断后,才会更新状态,也就是到最终状态的变更还是会有一点时间差

正确的代码编写

同步提交

思路:采用CountDownLatch,在spark程序未完成之前一直wait

public void syncSubmit(String mainClass, String args, Map<String, String> confMap) throws IOException {
	// 用于阻塞主线程,等待任务结束
	CountDownLatch countDownLatch = new CountDownLatch(1); 
    // 自定义SparkLauncher, 继承SparkLauncher
    sparkLauncher = new CustomSparkSubmitLauncher();
    // todo 吧啦吧啦 自己设置参数
    sparkLauncher
            .setJavaHome(xxx)
            .setSparkHome(xxx)
            .setMaster(xxx)
            .setDeployMode(xxx)
            .setAppName(xxx)
            .setVerbose(true);
    // 重点来了,启动spark程序,注册Listener,用于获取程序状态
    sparkAppHandle = sparkLauncher.startApplication(new SparkAppHandle.Listener() {
        @Override
        public void stateChanged(SparkAppHandle sparkAppHandle) {
            jobState = sparkAppHandle.getState();
            // print log when state change
            if (sparkAppHandle.getAppId() != null) {
                logger.info("{} stateChanged: {}", sparkAppHandle.getAppId(), jobState.toString());
            } else {
                logger.info("stateChanged: {}", jobState.toString());
            }
            // spark程序退出的错误判断条件
            //  if (jobState.isFinal()) {
            	//  countDownLatch.countDown();
            //  }
            // spark程序退出的正确判断条件
            if (sparkLauncher.isDisposed()){
                countDownLatch.countDown();
            }
        }
        @Override
        public void infoChanged(SparkAppHandle handle) {
        }
    });
    // 阻塞主线程
    countDownLatch.await();
}

异步提交

思路:提交后使用sparkLauncher.isDisposed()轮询监控spark程序是否退出

public void asyncSubmit(String mainClass, String args, Map<String, String> confMap) throws IOException {
    // 自定义SparkLauncher, 继承SparkLauncher
    sparkLauncher = new CustomSparkSubmitLauncher();
    // todo 吧啦吧啦 自己设置参数
    sparkLauncher
            .setJavaHome(xxx)
            .setSparkHome(xxx)
            .setMaster(xxx)
            .setDeployMode(xxx)
            .setAppName(xxx)
            .setVerbose(true);
    // 重点来了,启动spark程序,注册Listener,用于获取程序状态
    sparkAppHandle = sparkLauncher.startApplication(new SparkAppHandle.Listener() {
        @Override
        public void stateChanged(SparkAppHandle sparkAppHandle) {
            jobState = sparkAppHandle.getState();
            // print log when state change
            if (sparkAppHandle.getAppId() != null) {
                logger.info("{} stateChanged: {}", sparkAppHandle.getAppId(), jobState.toString());
            } else {
                logger.info("stateChanged: {}", jobState.toString());
            }
        }
        @Override
        public void infoChanged(SparkAppHandle handle) {
        }
    });
}

public void main() {
	asyncSubmit(参数)
	while (!sparkLauncher.isDisposed()) {
		Thread.sleep(5000)
		logger.info("job state: " + jobState)
	}
	logger.info("job final")
    // todo something
}

CustomSparkSubmitLauncher

注意:类名随意该类必须在org.apache.spark.launcher包下,所以在项目里新建一个相同路径即可
原因:AbstractAppHandle 不是public的,所以不能被其他包的类访问,为什么要用这个类,因为要用isDisposed()方法,那为什么要用这个方法呢,那就看下面的详细解释吧。文章来源地址https://uudwc.com/A/W1Bky

package org.apache.spark.launcher; // 包路径非常重要,必须是该路径

import java.io.IOException;


public class CustomSparkSubmitLauncher extends SparkLauncher {

    private SparkAppHandle sparkAppHandle;

    @Override
    public SparkAppHandle startApplication(SparkAppHandle.Listener... listeners) throws IOException {
        this.sparkAppHandle = super.startApplication(listeners);
        return sparkAppHandle;
    }

    public boolean isDisposed() {
        return ((AbstractAppHandle) sparkAppHandle).isDisposed();
    }
}

原文地址:https://blog.csdn.net/qq_37706484/article/details/129219283

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请联系站长进行投诉反馈,一经查实,立即删除!

h
上一篇 2023年09月17日 22:37
USB端口的ESD防护电路设计
下一篇 2023年09月17日 22:41