Structured Streaming怎么使用checkpoint進(jìn)行故障恢復(fù)

本篇文章給大家分享的是有關(guān)Structured Streaming怎么使用checkpoint進(jìn)行故障恢復(fù)，小編覺(jué)得挺實(shí)用的，因此分享給大家學(xué)習(xí)，希望大家閱讀完這篇文章后可以有所收獲，話不多說(shuō)，跟著小編一起來(lái)看看吧。

使用checkpoint進(jìn)行故障恢復(fù)

如果發(fā)生故障或關(guān)機(jī)，可以恢復(fù)之前的查詢的進(jìn)度和狀態(tài)，并從停止的地方繼續(xù)執(zhí)行。這是使用Checkpoint和預(yù)寫(xiě)日志完成的。您可以使用檢查點(diǎn)位置配置查詢，那么查詢將將所有進(jìn)度信息（即，每個(gè)觸發(fā)器中處理的偏移范圍）和運(yùn)行聚合（例如，示例中的wordcount）保存到檢查點(diǎn)位置。此檢查點(diǎn)位置必須是HDFS兼容文件系統(tǒng)中的路徑，并且可以在啟動(dòng)查詢時(shí)將其設(shè)置為DataStreamWriter中的選項(xiàng)。

aggDF
 .writeStream
 .outputMode("complete")
 .option("checkpointLocation", "path/to/HDFS/dir")
 .format("memory")
 .start()

具體測(cè)試代碼如下：

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進(jìn)行測(cè)試

以上就是Structured Streaming怎么使用checkpoint進(jìn)行故障恢復(fù)，小編相信有部分知識(shí)點(diǎn)可能是我們?nèi)粘９ぷ鲿?huì)見(jiàn)到或用到的。希望你能通過(guò)這篇文章學(xué)到更多知識(shí)。更多詳情敬請(qǐng)關(guān)注億速云行業(yè)資訊頻道。