使用checkpoint進行故障恢復

如果發生故障或關機，可以恢復之前的查詢的進度和狀態，並從停止的地方繼續執行。這是使用Checkpoint和預寫日誌完成的。您可以使用檢查點位置配置查詢，那麽查詢將將所有進度信息（即，每個觸發器中處理的偏移範圍）和運行聚合（例如，示例中的wordcount）保存到檢查點位置。此檢查點位置必須是HDFS兼容文件系統中的路徑，並且可以在啟動查詢時將其設置為DataStreamWriter中的選項。

具體測試代碼如下：

val lines =spark.readStream.format("socket").option("host", "localhost").option("port", 9999).load()

val words =lines.as[String].flatMap(_.split(" "))

val wordCounts =words.groupBy("value").count()
val query =wordCounts.writeStream.queryName("aggregates").outputMode("complete").option("checkpointLocation", "memory/").format("memory").start()

spark.sql("select * from aggregates").show()

kill掉submit進行測試

推薦閱讀：

1，2，StructuredStreaming的事件時間和窗口操作

2，譯：Tensorflow實現的CNN文本分類

3，機器學習之學習率 Learning Rate

3，Structured Streaming使用checkpoint進行故障恢復