从 Web UI 中杀死 Apache spark 作业并没有杀死它的 python 子进程答案

【问题标题】：Killing Apache spark job from web UI is not killing its python subprocess从 Web UI 中杀死 Apache spark 作业并没有杀死它的 python 子进程
【发布时间】：2019-02-24 08:32:42
【问题描述】：

pyspark 代码使用 subprocess.Popen(command) 调用另一个 python 作业

尝试从 spark master web UI http://localhost:8080 手动终止 Sparkcontext 并成功终止

当 python 子进程触发并作为 python 进程在工作节点中运行时。

使用 Redhatlinux

如果我杀死 pyspark sparkcontext，如何杀死 python 子进程？

【问题讨论】：

您是否对如何在事后杀死过时的进程，或者如何编写您的应用程序以使其自行清理有疑问？
如果作业成功或失败，我编写了代码来清理部分。寻找处理 sparkcontext 会话和 python 会话的解决方案。更多信息当前正在运行的 spark 作业由 spark API、驱动程序和应用程序触发在 spark master web UI 中显示。如果我杀死驱动程序，它也会从 UI 中自动杀死工作程序上的应用程序。但是，我从 spark JOB 触发的 python 子进程运行后台，我可以通过 shell 命令在工作程序节点中看到该进程，它赢了'不显示在从 UI loclhost:8081.I 需要如果我杀死 spark 应用程序驱动程序所有由 spark 作业触发的子进程触发器应该杀死

标签： python python-3.x apache-spark subprocess

【解决方案1】：

一般来说，可靠地终止子进程是相当困难的，因为当您想要终止子进程时，它可能正在执行不间断的代码。话虽如此，听起来“尽力而为”的方法可能适合您的情况。您将希望以一种允许您在进程中断时进行清理的方式创建并等待您的子进程。最简单的方法是将您的子进程放在 try/finally 块中。

try:
    print("starting subprocess")
    x = subprocess.Popen(["sleep", "100000"])
    x.wait()
finally:
    print("stopping subprocess")
    x.terminate()

我相信 spark 会发送中断信号

【讨论】：

嗨，Rico，感谢您的评论！我试过但没有用。
你能说得更具体点吗？没有到达finally块？调用了终止，但进程没有退出...您是否收到某种错误消息。
当我从 UI 中终止作业并且没有显示错误消息时，最终无法到达块。子进程仍在工作机器中运行。进程完成后显示的停止子进程消息

【解决方案2】：

下面的代码对我有用

from subprocess import Popen, PIPE, CalledProcessError
from contextlib import contextmanager
from pyspark import SparkContext
from pyspark import SparkConf
import sys, os, subprocess, signal, time

@contextmanager
def spark_manager():
    conf = SparkConf().setAppName("TEST-SPARK")
    conf.set("spark.scheduler.mode", "FAIR")
    sc = SparkContext(conf=conf)

    try:
        yield sc
    finally:
        sc.stop()

with spark_manager() as context:
    process = subprocess.Popen(['python3', 'test.py'], shell=False, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)

    # Poll process for new output until finished
    while True:
        if context._jsc.sc().isStopped():
            print(process.pid)
            time.sleep(1.0)
            os.kill(process.pid, signal.SIGKILL)
            break
        nextline = process.stdout.readline()
        if nextline == '' and process.poll() is not None:
            break
        sys.stdout.write(nextline)
        sys.stdout.flush()

    output = process.communicate()[0]
    exitCode = process.returncode

    if (exitCode == 0):
        print(output)
    else:
        raise ProcessException(command, exitCode, output)

【讨论】：