如何从spark-submit获取返回码？

Question

我想在远程位置使用ssh连接执行一个spark作业。

有一些情况下，作业失败，但调度程序将其标记为“成功”，所以我想检查火花提交的返回代码，所以我可以强制失败它。

下面是我使用

def execute_XXXX(): 
    f = open('linux.pem','r') 
    s = f.read() 
    keyfile = StringIO.StringIO(s) 
    mykey = paramiko.RSAKey.from_private_key(keyfile) 
    sshcon = paramiko.SSHClient() 
    sshcon.set_missing_host_key_policy(paramiko.AutoAddPolicy()) 
    sshcon.connect('XXX.XXX.XXX.XXX', username='XXX', pkey=mykey) 
    stdin, stderr, stdout= sshcon.exec_command('spark-submit XXX.py') 

    logger.info("XXX ------>"+str(stdout.readlines())) 
    logger.info("Error--------->"+str(stderr.readlines())) 

代码如何获得的返回码火花提交的工作，所以我可以有力地失败的任务。 或者你可以建议一个替代解决方案。

感谢，阿赫亚

Answer 1

所以这是我如何解决我所面临的问题。简单的1行代码就足够了。

def execute_XXXX(): 
    f = open('linux.pem','r') 
    s = f.read() 
    keyfile = StringIO.StringIO(s) 
    mykey = paramiko.RSAKey.from_private_key(keyfile) 
    sshcon = paramiko.SSHClient() 
    sshcon.set_missing_host_key_policy(paramiko.AutoAddPolicy()) 
    sshcon.connect('XXX.XXX.XXX.XXX', username='XXX', pkey=mykey) 
    stdin, stderr, stdout= sshcon.exec_command('spark-submit XXX.py') 
    if (stdout.channel.recv_exit_status())!= 0: 
     logger.info("XXX ------>"+str(stdout.readlines())) 
     logger.info("Error--------->"+str(stderr.readlines())) 
     sys.exit(1) 

Answer 2

你需要实现一个sparkListener。更多信息可以在下面的链接找到。

How to implement custom job listener/tracker in Spark?