spark终止运行_如何优雅地终止正在运行的SparkStreaming程序

一直运行的Streaming程序如何关闭呢&＃xff1f;是直接使用kill命令强制关闭吗&＃xff1f;这种手段是可以达到关闭的目的&＃xff0c;但是带来的后果就是可能会导致数据的丢失&＃xff0c;因为这时候如果程序正在处理接收到的数据&＃xff0c;但是由于接收到kill命令&＃xff0c;那它只能停止整个程序&＃xff0c;而那些正在处理或者还没有处理的数据可能就会被丢失。那我们咋办&＃xff1f;这里有两种方法。

文章目录

等作业运行完再关闭

我们都知道&＃xff0c;Streaming每隔batchDuration的时间会把源源不断的流数据分割成一批有限数据集&＃xff0c;然后计算这些数据&＃xff0c;我们可以从提供的监控页面看到当前batch是否执行完成&＃xff0c;当作业执行完&＃xff0c;我们就可以手动执行kill命令来强制关闭这个Streaming作业。这种方式的缺点就是得盯着监控页面&＃xff0c;然后决定关不关闭&＃xff0c;很不灵活。

如果想及时了解Spark、Hadoop或者Hbase相关的文章&＃xff0c;欢迎关注微信公共帐号&＃xff1a;iteblog_hadoop

通过Spark内置机制关闭

其实Spark内置就为我们提供了一种优雅的方法来关闭长期运行的Streaming作业&＃xff0c;我们来看看 StreamingContext类中定义的一个 stop 方法&＃xff1a;

def stop(stopSparkContext : Boolean, stopGracefully : Boolean)

官方文档对其解释是&＃xff1a;Stop the execution of the streams, with option of ensuring all received data has been processed.控制所有接收的数据是否被处理的参数就是 stopGracefully&＃xff0c;如果我们将它设置为true&＃xff0c;Spark则会等待所有接收的数据被处理完成&＃xff0c;然后再关闭计算引擎&＃xff0c;这样就可以避免数据的丢失。现在的问题是我们在哪里调用这个stop方法&＃xff1f;

Spark 1.4版本之前

在Spark 1.4版本之前&＃xff0c;我们需要手动调用这个 stop 方法&＃xff0c;一种比较合适的方式是通过 Runtime.getRuntime().addShutdownHook 来添加一个钩子&＃xff0c;其会在JVM关闭的之前执行传递给他的函数&＃xff0c;如下&＃xff1a;

Runtime.getRuntime().addShutdownHook( new Thread() {

override def run() {

log( "Gracefully stop Spark Streaming" )

streamingContext.stop( true , true )

}

})

如果你使用的是Scala&＃xff0c;我们还可以通过以下的方法实现类似的功能&＃xff1a;

scala.sys.addShutdownHook({

streamingContext.stop( true , true )

)})

通过上面的办法&＃xff0c;我们客户确保程序退出之前会执行上面的函数&＃xff0c;从而保证Streaming程序关闭的时候不丢失数据。

Spark 1.4版本之后

上面方式可以达到我们的需求&＃xff0c;但是在每个程序里面都添加这样的重复代码也未免太过麻烦了&＃xff01;值得高兴的是&＃xff0c;从Apache Spark 1.4版本开始&＃xff0c;Spark内置提供了spark.streaming.stopGracefullyOnShutdown参数来决定是否需要以Gracefully方式来关闭Streaming程序(详情请参见)。Spark会在启动 StreamingContext 的时候注册这个钩子&＃xff0c;如下&＃xff1a;

shutdownHookRef &＃61; ShutdownHookManager.addShutdownHook(

StreamingContext.SHUTDOWN _ HOOK _ PRIORITY)(stopOnShutdown)

private def stopOnShutdown() : Unit &＃61; {

val stopGracefully &＃61; conf.getBoolean( "spark.streaming.stopGracefullyOnShutdown" , false )

logInfo(s "Invoking stop(stopGracefully&＃61;$stopGracefully) from shutdown hook" )

// Do not stop SparkContext, let its own shutdown hook stop it

stop(stopSparkContext &＃61; false , stopGracefully &＃61; stopGracefully)

}

从上面的代码可以看出&＃xff0c;我们可以根据自己的需求来设置 spark.streaming.stopGracefullyOnShutdown 的值&＃xff0c;而不需要在每个Streaming程序里面手动调用StreamingContext的stop方法&＃xff0c;确实方便多了。不过虽然这个参数在Spark 1.4开始引入&＃xff0c;但是却是在Spark 1.6才开始才有文档正式介绍(可以参见https://github.com/apache/spark/pull/8898和http://spark.apache.org/docs/1.6.0/configuration.html)

猜你喜欢

欢迎关注本公众号&＃xff1a;iteblog_hadoop:

0、回复spark_2017_all获取Spark Summit East 2017高清视频和PPT

1、回复hive_es获取《基于 Hive/ES 金融大数据指标系统》PPT

2、回复bigdata_e获取 《大规模数据处理演变》PPT

3、回复 大数据分析获取 《Big Data Analytics》电子书

4、回复 spark2电子书获取 《Apache Spark 2 for Beginners》电子书

5、回复 spark2_data获取 《Spark for Data Science》电子书

6、回复 架构师大会ppt获取 《2016年中国架构师[大数据场]》PPT

7、回复 intro_flink获取 《Introduction to Apache Flink》电子书

8、回复learning_flink获取《Learning Apache Flink》电子书

8、回复Hadoop权威指南获取 《Hadoop权威指南中文第三版》电子书

9、回复 flink未来获取 《The Future of Apache Flink》

10、更多大数据文章欢迎访问https://www.iteblog.com及本公众号(iteblog_hadoop)返回搜狐&＃xff0c;查看更多