由於streaming流程序一旦運行起來，基本上是無休止的狀態，除非是特殊情況，否則是不會停的。因為每時每刻都有可能在處理數據，如果要停止也需要確認當前正在處理的數據執行完畢，並且不能再接受新的數據，這樣才能保證數據不丟不重。
同時，也由於流程序比較特殊，所以也不能直接kill -9這種暴力方式停掉，直接kill的話，就有可能丟失數據或者重復消費數據。
下面介紹如何優雅的停止streaming job。
第一種：人工手動停止
- 程序里設置如下參數：
sparkConf.set("spark.streaming.stopGracefullyOnShutdown","true")//優雅的關閉
- 然后按照下面步驟操作
- 通過Hadoop 8088頁面找到運行的程序
- 打開spark UI的監控頁面
- 打開executor的監控頁面
- 登錄Linux找到驅動節點所在的機器IP以及運行的端口號
- 然后執行一個封裝好的命令
sudo ss -tanlp | grep 5555 |awk '{print $6}'|awk -F, '{print $2}' | sudo xargs kill -15
這種方式顯然是比較復雜的。
第二種：使用HDFS系統做消息通知
在驅動程序中，加上一段代碼，作用就是每隔一段時間掃描HDFS上一個文件，如果發現這個文件存在，就調用StreamContext的Stop方法，優雅的停止程序。
這里的HDFS可以換成reids、zk、hbase、db，唯一的問題就是依賴了外部的一個存儲系統來達到消息通知的目的。
使用這種方式，停止程序就比較簡單。登錄有HDFS客戶端的機器，然后touch一個空文件到指定目錄，等到間隔的掃描時間，發現有文件存在，就需要關閉程序了。
廢話不多說，上代碼
ssc.start()
//check interval
val checkIntervalMillis = 15000
var isStopped = false
println("before while")
while (!isStopped) {
println("calling awaitTerminationOrTimeout")
isStopped = ssc.awaitTerminationOrTimeout(checkIntervalMillis)
if (isStopped)
println("confirmed! The streaming context is stopped. Exiting application...")
else
println("Streaming App is still running.")
println("check file exists")
if (!stopFlag) {
val fs = FileSystem.get(new URI("hdfs://192.168.156.111:9000"),new Configuration())
stopFlag = fs.exists(new Path("/stopMarker/marker"))
}
if (!isStopped && stopFlag) {
println("stopping ssc right now")
ssc.stop(true, true)
}
}
第三種：內部暴露一個socket或者http端口用來接收請求，等待除法關閉流程序
這種方式需要在driver啟動一個socket線程，或者http服務。比較推薦使用http服務，因為socket有點偏底層，處理起來稍微復雜。
如果使用http服務，可以直接用內嵌的jetty，對外暴露一個http接口。Spark UI頁面用的也是內嵌的jetty提供服務，所以不需要在pom文件引入額外的依賴，在關閉的時候，找到驅動所在的IP，就可以直接通過crul或者瀏覽器直接關閉流程序
找到驅動程序所在的IP，可以在程序啟動的log中看到，也可以在spark master UI界面上找到，這種方式不依賴任何的存儲系統，僅僅在部署的時候需要一個額外的端口號暴露http服務。
推薦使用第二種或第三種，如果想最大程度的減少對外部系統的依賴，推薦使用第三種。
參考文檔：https://www.linkedin.com/pulse/how-shutdown-spark-streaming-job-gracefully-lan-jiang