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

原文发布于微信公众号 - Spark学习技巧(bigdatatip)

原文发表时间:2018-05-13

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏Hadoop实操

2.如何在RedHat7中实现OpenLDAP集成SSH登录并使用sssd同步用户

温馨提示:要看高清无码套图,请使用手机打开并单击图片放大查看。 Fayson的github:https://github.com/fayson/cdhproje...

2.4K9
来自专栏菩提树下的杨过

hadoop: hive 1.2.0 在mac机上的安装与配置

环境:mac OS X Yosemite + hadoop 2.6.0 + hive 1.2.0 + jdk 1.7.0_79 前提:hadoop必须先安装,且...

2668
来自专栏个人分享

spark MapOutputTrackerMaster

最近用了一个RowNumber() over()函数 进行三张4000万数据的关联筛选,建表语句如下:

1082
来自专栏分布式系统进阶

Kafka是如何处理客户端发送的数据的?

// respond immediately if // 1) fetch request does not w...

1091
来自专栏星汉技术

HBase简介及搭建

3634
来自专栏xingoo, 一个梦想做发明家的程序员

Spark监控官方文档学习笔记

任务的监控和使用 有几种方式监控spark应用:Web UI,指标和外部方法 Web接口 每个SparkContext都会启动一个web UI,默认是40...

2929
来自专栏Hadoop实操

如何实现CDH元数据库MySQL的主主互备

在前面Fayson讲过《如何实现CDH元数据库MySQL的主备》,而本篇文章介绍如何实现MySQL的双活方式,为后面基于Keepalived实现MySQL高可用...

4668
来自专栏Hadoop实操

重庆某项目生产集群扩容问题总结及复盘

本文主要讲述重庆某项目生产集群扩容项目问题总结及复盘。其中部分问题之前有写过相关文档,可参考我之前写的文章《CDH集群安装YARN无法正常启动及解决办法》、《H...

1431
来自专栏杂烩

kafka-eagle1.1.9安装 原

源码地址https://github.com/smartloli/kafka-eagle

851
来自专栏FreeBuf

利用PowerShell Empire实现Word文档DDE攻击控制

? 近期,有安全人员发现了一种DynamicData Exchange (DDE)协议绕过MSWord和MSExcel宏限制,不需使用MSWord和MSExc...

2408

扫码关注云+社区