[spark streaming]Executor容错安全性

谁借莪１个温暖的怀抱￠ 2022-05-14 00:29 218阅读 0赞

**一、Spark Streaming 数据安全性的考虑：**

1.  Spark Streaming不断的接收数据，并且不断的产生Job，不断的提交Job给集群运行。所以这就涉及到一个非常重要的问题数据安全性。
2.  Spark Streaming是基于Spark Core之上的，如果能够确保数据安全可好的话，在Spark Streaming生成Job的时候里面是基于RDD，即使运行的时候出现问题，那么Spark Streaming也可以借助Spark Core的容错机制自动容错。
3.  对Executor容错主要是对数据的安全容错
4.  为啥这里不考虑对数据计算的容错：计算的时候Spark Streaming是借助于Spark Core之上的容错的，所以天然就是安全可靠的。

**Executor容错方式：**   
1. 最简单的容错是副本方式，基于底层BlockManager副本容错，也是默认的容错方式。

2.WAL日志方式

3. 接收到数据之后不做副本，支持数据重放，所谓重放就是支持反复读取数据。

**BlockManager备份：**

1.  默认在内存中两份副本，也就是Spark Streaming的Receiver接收到数据之后存储的时候指定StorageLevel为MEMORY\_AND\_DISK\_SER\_2，底层存储是交给BlockManager，BlockManager的语义确保了如果指定了两份副本，一般都在内存中。所以至少两个Executor中都会有数据。

![592648-20160530095248586-578465280.png][]

Receiver将数据交给BlockManger是由ReceiveredBlockHandler来处理的，有两种ReceiveredBlockHandler的实现：

1.WriteAheadLogBasedBlockHandler

2.BlockManagerBasedBlockHandler

![592648-20160530095249211-349335055.png][]

这里的storageLevel是构建InputDStream时传入的，socketTextStream的默认存储级别是StorageLevel.MEMORY\_AND\_DISK\_SER\_2

![592648-20160530095249930-632169309.png][]

如果使用WriteAheadLogBasedBlockHandler需要开启WAL，默认并没有开启：

![592648-20160530095250524-2070926863.png][]

** **

**WAL日志方式：**

这种方式会现将数据写入日志文件，就是checkpoint目录，出现异常是，从checkpoint目录重新读取数据，进行恢复。启动WAL时候，没必要将副本数设置成大于1，也不需要序列化。

![592648-20160530095250946-1122961332.png][]

WAL会将数据同时写入BlockManager和write ahead log，而且是并行的写block，当然两处的block存储完成，才会返回。

![592648-20160530095251352-1019828628.png][]

将Block 存入BlockManager：

![592648-20160530095251883-2145644157.png][]

将Block 存入WAL日志：

![592648-20160530095252352-1169549061.png][]

WAL写数据的时候是顺序写，数据不可修改，所以读的时候只需要按照指针(也就是要读的record在那，长度是多少)读即可。所以WAL的速度非常快。

浏览一下WriteAheadLog，他是一个抽象类：

看一下WriteAheadLog的一个实现类FileBasedWriteAheadLog的write方法：

根据不同时间获取不同Writer将序列化结果写入文件,返回一个FileBasedWriteAheadLogSegment类型的对象fileSegment。

![592648-20160530095253399-132478812.png][]

读数据：

![592648-20160530095253789-1241557088.png][]

其中创建了一个FileBaseWriteAheadLogRandomReader对象，然后调用了该对象的read方法:

![592648-20160530095254305-488408394.png][]

**支持数据重放。**

在实际的开发中直接使用Kafka，因为不需要容错，也不需要副本。   
Kafka有Receiver方式和Direct方式   
Receiver方式：是交给Zookeeper去管理数据的，也就是偏移量offSet.如果失效后，Kafka会基于offSet重新读取，因为处理数据的时候中途崩溃，不会给Zookeeper发送ACK，此时Zookeeper认为你并没有消息这个数据。但是在实际中越来用的越多的是Direct的方式直接操作offSet.而且还是自己管理offSet.

1.  DirectKafkaInputDStream会去查看最新的offSet,并且把offSet放到Batch中。
2.  在Batch每次生成的时候都会调用latestLeaderOffsets查看最近的offSet,此时的offSet就会与上一个offSet相减获得这个Batch的范围。这样就可以知道读那些数据。

![592648-20160530095254946-1823049668.png][]

protected final def latestLeaderOffsets(retries: Int): Map[TopicAndPartition, LeaderOffset] = {
      val o = kc.getLatestLeaderOffsets(currentOffsets.keySet)
      // Either.fold would confuse @tailrec, do it manuallyif (o.isLeft) {
        val err = o.left.get.toString
        if (retries <= 0) {
          throw new SparkException(err)
        } else {
          log.error(err)
          Thread.sleep(kc.config.refreshLeaderBackoffMs)
          latestLeaderOffsets(retries - 1)
        }
      } else {
        o.right.get
      }
    }

[592648-20160530095248586-578465280.png]: /images/20220514/b82f2298a11649629f4b026ee748b797.png
[592648-20160530095249211-349335055.png]: /images/20220514/622ee8a802ae49fab8e5a43af27e8560.png
[592648-20160530095249930-632169309.png]: /images/20220514/82880cb8e6fa4a958287a07c2093fefa.png
[592648-20160530095250524-2070926863.png]: /images/20220514/02ca50a273d54a5ab2dc1601b5475931.png
[592648-20160530095250946-1122961332.png]: /images/20220514/c9e2276396424ee18eb0df560a43e4b2.png
[592648-20160530095251352-1019828628.png]: /images/20220514/14b8627f941f4d99b333bed00c46cf92.png
[592648-20160530095251883-2145644157.png]: /images/20220514/beddb7468ff44c039744390d2aff4865.png
[592648-20160530095252352-1169549061.png]: /images/20220514/798772357fb84c0bb83f6cad44fd9f1d.png
[592648-20160530095253399-132478812.png]: /images/20220514/2ef9bd5f27d84c96ae3552538bcebd20.png
[592648-20160530095253789-1241557088.png]: /images/20220514/5bfc407355e04938aee9a881b372dcd8.png
[592648-20160530095254305-488408394.png]: /images/20220514/5bddd0705271484ebe42ec9ff97ddb27.png
[592648-20160530095254946-1823049668.png]: /images/20220514/cd0b14846a774d459b9104c26980d8d2.png