StructuredStreaming怎么使用checkpoint进行故障恢复

本篇文章给大家分享的是有关Structured Streaming怎么使用checkpoint进行故障恢复，小编觉得挺实用的，因此分享给大家学习，希望大家阅读完这篇文章后可以有所收获，话不多说，跟着小编一起来看看吧。

创新互联公司服务项目包括泉州网站建设、泉州网站制作、泉州网页制作以及泉州网络营销策划等。多年来，我们专注于互联网行业，利用自身积累的技术优势、行业经验、深度合作伙伴关系等，向广大中小型企业、政府机构等提供互联网行业的解决方案，泉州网站推广取得了明显的社会效益与经济效益。目前，我们服务的客户以成都为中心已经辐射到泉州省份的部分城市，未来相信会继续扩大服务区域并继续获得客户的支持与信任！

使用checkpoint进行故障恢复

如果发生故障或关机，可以恢复之前的查询的进度和状态，并从停止的地方继续执行。这是使用Checkpoint和预写日志完成的。您可以使用检查点位置配置查询，那么查询将将所有进度信息（即，每个触发器中处理的偏移范围）和运行聚合（例如，示例中的wordcount）保存到检查点位置。此检查点位置必须是HDFS兼容文件系统中的路径，并且可以在启动查询时将其设置为DataStreamWriter中的选项。

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

具体测试代码如下：

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进行测试

以上就是Structured Streaming怎么使用checkpoint进行故障恢复，小编相信有部分知识点可能是我们日常工作会见到或用到的。希望你能通过这篇文章学到更多知识。更多详情敬请关注创新互联行业资讯频道。