spark checkpoint用法

原创
2016/09/02 09:41
阅读数 6.8K

checkpoint注意点:

1.当某一窗口被打断之后,重新从checkpoint恢复时,不会继续执行这一窗口未执行完的代码,仅仅是恢复spark streaming的配置和代码,进行下一批次的执行。

2.看到某些文章说spark streaming重新编译了之后,不能从checkpoint取出数据,继续执行,但是我运行代码检测到,即使重新打包,也是可以继续从checkpoint取出数据执行的。
3.报map什么没有初始化的错误原因是:dstream的api需要写在createStreamingContext方法里面,不能写在StreamingContext.getOrCreate这一行代码的后面,否则就会报刚刚那个错误了。

4.ssc.checkpoint(path),如果放在集群上跑,则path默认是hdfs目录,如果在本地运行,则path是本地目录。

5.metadata checkpoint有一点不好理解:保存那些在队列中还没有处理的batch(仅仅保存元数据,而不是这些batch中的数据),按照我的理解就是,recevier接收数据后,如果spark streaming处理不过来(比如说设置了没一次最多多少条数据),那么就会先将数据按批次放到队列里,当这个时候如果driver program挂了,那么为开始的队列就会被保存起来,等下次启动任务的时候,继续执行那些未被执行的队列。如果正在执行的队列被打断,下次启动任务的时候,应该就不会继续执行正好被打断的批次了。仔细研读:Spark Streaming和Kafka整合是如何保证数据零丢失,里面的“可能存在数据丢失的场景”验证我上面的理解是正确的。

6.总结就是,checkpoint不能作为spark streaming保证数据可靠性的解决方案,需要我们自己用代码实现来保证数据的可靠处理。

测试代码:

package com.bigdata.spark.streaming.examples
import org.apache.spark._
import org.apache.spark.streaming._
import org.apache.spark.streaming.StreamingContext._
import org.apache.hadoop.util.ThreadUtil
import com.bigdata.util._
import java.util.Date
object TestCheckpoint {
  def createStreamingContext(): StreamingContext = {
    val conf = new SparkConf().setAppName("NetworkWordCount")
    conf.setMaster("local[4]")
    val ssc = new StreamingContext(conf, Seconds(5))
    ssc.checkpoint("D://checkpoint")
    val lines = ssc.socketTextStream("192.168.1.11", 9999)
    val words = lines.flatMap(_.split(" "))
    val wa = words.map(a).persist()
    val wb = wa.map(b)
    val pairs = words.map(word => (word, 1))
    val wordCounts = pairs.reduceByKey(_ + _)
    wb.print()
    return ssc
  }
  def main(args: Array[String]): Unit = {
    val ssc = StreamingContext.getOrCreate("/tmp/checkpoint", createStreamingContext)
    ssc.start()             // Start the computation
    ssc.awaitTermination()  // Wait for the computation to terminate
  }
   def a(line: String): String = {
      val re = line + " a "
      val hdfsUtil = new HDFSUtil()
      hdfsUtil.createFile("/tmp/a" + new Date().getTime, re.toString)
    return re
  }
  def b(line: String) = {
      Thread.sleep(10000)
      val re = line + " b "
      val hdfsUtil = new HDFSUtil()
      hdfsUtil.createFile("/tmp/b"  + new Date().getTime, re.toString)
  }
}

 

checkpoint分类

(1)Metadata checkpointing
将流式计算的信息保存到具备容错性的存储上如HDFS,Metadata Checkpointing适用于当streaming应用程序Driver所在的节点出错时能够恢复,元数据包括:
Configuration(配置信息) - 创建streaming应用程序的配置信息
DStream operations - 在streaming应用程序中定义的DStreaming操作
Incomplete batches - 在列队中没有处理完的作业

(2)Data checkpointing
将生成的RDD保存到外部可靠的存储当中,对于一些数据跨度为多个bactch的有状态tranformation操作来说,checkpoint非常有必要,因为在这些transformation操作生成的RDD对前一RDD有依赖,随着时间的增加,依赖链可能会非常长,checkpoint机制能够切断依赖链,将中间的RDD周期性地checkpoint到可靠存储当中,从而在出错时可以直接从checkpoint点恢复。

具体来说,metadata checkpointing主要还是从drvier失败中恢复,而Data Checkpoing用于对有状态的transformation操作进行checkpointing

参考:Spark修炼之道(进阶篇)——Spark入门到精通:第十四节 Spark Streaming 缓存、Checkpoint机制

展开阅读全文
加载中
点击引领话题📣 发布并加入讨论🔥
打赏
0 评论
0 收藏
0
分享
返回顶部
顶部