如何优雅地停止 YARN 上的 Spark Streaming 应用程序？

Question

我正在集群模式下在 YARN 上运行 Spark Streaming 应用程序，我正在尝试实现正常关闭，以便在应用程序被终止时，它将在停止之前完成当前微批处理的执行。

按照一些教程，我将spark.streaming.stopGracefullyOnShutdown 配置为true，并将以下代码添加到我的应用程序中：

sys.ShutdownHookThread {
   log.info("Gracefully stopping Spark Streaming Application")  
   ssc.stop(true, true)
   log.info("Application stopped")
}

但是当我用

杀死应用程序时

yarn application -kill application_1454432703118_3558

此时执行的微批处理没有完成。

在驱动程序中，我看到打印的第一行日志（“正常停止 Spark Streaming 应用程序”），但没有看到最后一行（“应用程序停止”）。

ERROR yarn.ApplicationMaster: RECEIVED SIGNAL 15: SIGTERM
INFO streaming.MySparkJob: Gracefully stopping Spark Streaming Application
INFO scheduler.JobGenerator: Stopping JobGenerator gracefully
INFO scheduler.JobGenerator: Waiting for all received blocks to be consumed for job generation
INFO scheduler.JobGenerator: Waited for all received blocks to be consumed for job generation
INFO streaming.StreamingContext: Invoking stop(stopGracefully=true) from shutdown hook

在 executors 日志中，我看到以下错误：

ERROR executor.CoarseGrainedExecutorBackend: Driver 192.168.6.21:49767 disassociated! Shutting down.
INFO storage.DiskBlockManager: Shutdown hook called
WARN remote.ReliableDeliverySupervisor: Association with remote system [akka.tcp://sparkDriver@192.168.6.21:49767] has failed, address is now gated for [5000] ms. Reason: [Disassociated]
INFO util.ShutdownHookManager: Shutdown hook called

我认为问题与 YARN 如何向应用程序发送终止信号有关。关于如何使应用程序正常停止的任何想法？

Answer 1

你应该去 executors 页面看看你的驱动程序在哪里运行（在哪个节点上）。 ssh 到该节点并执行以下操作：

ps -ef | grep 'app_name'

（将 app_name 替换为您的类名/应用程序名）。它将列出几个进程。看看这个过程，有些会是另一个的孩子。选择最父进程的 id 并发送一个 SIGTERM

kill pid

一段时间后，您会看到您的应用程序已正常终止。

而且现在您不需要添加这些挂钩来关闭。 使用spark.streaming.stopGracefullyOnShutdown 配置帮助正常关机

Answer 2

您可以通过在触发自定义条件时调用ssc.stop 而不是使用awaitTermination 来停止火花流应用程序。如下伪代码所示：

ssc.start()
while True:
    time.sleep(10s)
    if some_file_exist:
        ssc.stop(True, True)