如何從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?