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 条评论
登录 后参与评论

相关文章

来自专栏分布式系统进阶

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

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

861
来自专栏禅林阆苑

利用NextCloud配置私有云 【原创】

利用NextCloud配置私有云 Write By CS逍遥剑仙 我的主页: www.csxiaoyao.com GitHub: github...

1.2K8
来自专栏个人分享

spark MapOutputTrackerMaster

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

972
来自专栏琯琯博客

开发 Composer 包详细步骤

一、GitHub 创建一个名 uploadfile 新仓库,并克隆至本地。 二、初始化项目,生成composer.json文件 2.1 步骤 2.2 步骤解释...

61212
来自专栏跟着阿笨一起玩NET

Log4Net日志记录两种方式

     log4net库是Apache log4j框架在Microsoft .NET平台的实现,是一个帮助程序员将日志信息输出到各种目标(控制台、文件、数据库...

792
来自专栏码字搬砖

Hbase原理详解

首先指出图片的一个错误,Hlog应该属于HRegionserver的,不应该在HRegion中。

1573
来自专栏FreeBuf

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

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

2318
来自专栏Hadoop实操

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

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

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

Spark监控官方文档学习笔记

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

2619
来自专栏杂烩

kafka-eagle1.1.9安装 原

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

641

扫码关注云+社区