Checkpointing

一个流应用程序必须全天候运行，所有必须能够解决应用程序逻辑无关的故障（如系统错误，JVM崩溃等）。为了使这成为可能，Spark Streaming需要checkpoint足够的信息到容错存储系统中，以使系统从故障中恢复。

Metadata checkpointing：保存流计算的定义信息到容错存储系统如HDFS中。这用来恢复应用程序中运行worker的节点的故障。元数据包括
- Configuration ：创建Spark Streaming应用程序的配置信息
- DStream operations ：定义Streaming应用程序的操作集合
- Incomplete batches：操作存在队列中的未完成的批
Data checkpointing ：保存生成的RDD到可靠的存储系统中，这在有状态transformation（如结合跨多个批次的数据）中是必须的。在这样一个transformation中，生成的RDD依赖于之前批的RDD，随着时间的推移，这个依赖链的长度会持续增长。在恢复的过程中，为了避免这种无限增长。有状态的transformation的中间RDD将会定时地存储到可靠存储系统中，以截断这个依赖链。

元数据checkpoint主要是为了从driver故障中恢复数据。如果transformation操作被用到了，数据checkpoint即使在简单的操作中都是必须的。

Metadata checkpointing

相关代码：

def createContext(checkpointDirectory: String)
: StreamingContext = {// If you do not see this printed, that means the StreamingContext has been loaded// from the new checkpointprintln("Creating new context")val sparkConf = new SparkConf().setAppName("DynamicRange")// Create the context with a 1 second batch sizeval ssc = new StreamingContext(sparkConf, Seconds(8))ssc.checkpoint(checkpointDirectory)...//你的kafka streaming的相关代码最好放在这里，不然有可能抛异常：spark checkpoint KafkaInputDStream has not been initialize//create kafka streamval fullLines = KafkaUtils.createStream(ssc, SystemConfig.config.kafkaZkQuorum, SystemConfig.config.kafkaGroup, topicMap);//parse data stringval valueLines = fullLines.map(_._2)..ssc
}def main(args: Array[String]) {var ssc: StreamingContext = nulltry {ssc = StreamingContext.getOrCreate(".", () => {println("get context fail, try to create a new one.")createContext(".")})} catch{case e:Exception =>{println("get context exception, try to create a new one.")ssc = createContext(".")}}ssc.start()ssc.awaitTermination()
}

注意：

kafka streaming的相关代码最好放在createContext里面，不然有可能抛异常：spark checkpoint KafkaInputDStream has not been initialize。
不同版本之间的Spark Driver是不能从文件中恢复的，所以这里我用try catch如果有异常就新建一个context。

转载于:https://www.cnblogs.com/hd-zg/p/5831235.html

Spark Streaming metadata checkpoint相关推荐

Spark Streaming之checkpoint机制
一什么类型的数据需要使用checkpoint? Spark Streaming是最需要进行容错的,因为一般都是7 * 24小时运转,所以需要将足够的信息checkpoint到容错的存储系统上,比如H ...
spark基础之spark streaming的checkpoint机制
一什么类型的数据需要使用checkpoint? Spark Streaming是最需要进行容错的,因为一般都是7 * 24小时运转,所以需要将足够的信息checkpoint到容错的存储系统上,比如H ...
Spark Streaming 实战案例（四) Spark Streaming 缓存、Checkpoint机制
主要内容 Spark Stream 缓存 Checkpoint 案例 1. Spark Stream 缓存通过前面一系列的课程介绍,我们知道DStream是由一系列的RDD构成的,它同一般的RDD一 ...
spark streaming checkpointing
spark streaming的相关概念: spark的核心是创建一个RDD对象,然后对RDD对象进行计算操作等 streaming可以理解为是一个连续不断的数据流 ,然后将每个固定时间段里的数据构 ...
Spark Streaming学习与实践(2)
2. 运算 2.1. StreamingContext.checkPoint val ssc = new StreamingContext(sc, Seconds(10))val checkpoint ...
Spark Streaming笔记整理（二）：案例、SSC、数据源与自定义Receiver
[TOC] 实时WordCount案例主要是监听网络端口中的数据,并实时进行wc的计算. Java版测试代码如下: package cn.xpleaf.bigdata.spark.java.str ...
Spark Streaming使用Kafka保证数据零丢失
为什么80%的码农都做不了架构师?>>> 源文件放在github,随着理解的深入,不断更新,如有谬误之处,欢迎指正.原文链接https://github.com/jacksu/ ...
spark streaming 接收 kafka 数据java代码WordCount示例
1. 首先启动zookeeper 2. 启动kafka 3. 核心代码生产者生产消息的java代码,生成要统计的单词 package streaming;import java.util.Prope ...
Spark Streaming学习笔记
特点: Spark Streaming能够实现对实时数据流的流式处理,并具有很好的可扩展性.高吞吐量和容错性. Spark Streaming支持从多种数据源提取数据,如:Kafka.Flume.Tw ...

Spark Streaming metadata checkpoint

Checkpointing

Metadata checkpointing

相关代码：

注意：

Spark Streaming metadata checkpoint相关推荐

最新文章

热门文章