SparkStreaming 性能、稳定、容错与语义

提示：扫一扫查出行
【扫一扫了解最新限行尾号】
复制提示

怎样提高Spark Streaming的性能

1、创建多个接收器

待定：：

2、调节每一个batch interval的数据块的数量，其实就是调整上面第二个问题中提到的配置spark.streaming.blockInterva

待定：：

3、调整Recevier每秒接收数据的速率

待定：：

4、通过repartition这个API来增加并行度

待定：：

5、使用Kryo序列化机制

待定：：

6、使用CMS垃圾收集器

待定：：

Spark Streaming的稳定性：

BackPressure

待定：：

Elastic Scaling

待定：：

Spark Streaming是怎样容错的

1、Executor失败容错：Executor的失败会重新启动一个新的Executor，这个是Spark自身的特性。如果Receiver所在的Executor失败了，那么Spark Streaming会在另外一个Executor上启动这个Receiver(这个Executor上可能存在已经接收到的数据的备份)

2、Driver失败的容错：如果Driver失败的话，那么整个Spark Streaming应用将会全部挂掉。所以Driver端的容错是非常重要的，我们首先可以配置Driver端的checkpoint，用于定期的保存Driver端的状态；然后我们可以配置Driver端失败的自动重启机制(每一种集群管理的配置都不一样)；最后我们需要打开Executor端的WAL机制

3、一个Task失败的容错：Spark中的某个Task失败了可以重新运行，这个Task所在的Stage失败的话呢，也可以根据RDD的依赖重新跑这个Stage的父亲Stage，进而重新跑这个失败的Stage

4、在实时计算的过程，肯定不能容忍某个Task的运行时间过长，Spark Streaming对于某个运行时间过长的Task会将这个Task杀掉重新在另一个资源比较充足的Executor上执行。这个就是利用了Spark的Task调度的推测机制。

Spark Streaming程序怎么做到不丢数据

答：因为Spark Streaming在接收数据的时候有两种模式，第一种是基于Receiver模式，第二种是Kafka Direct模式，两者不丢数据的处理方式不一样，所以我们需要了解掌握这两种模式不丢数据的处理策略：

基于Receiver模式：

在这种模式下，我们可以使用checkpoint + WAL + ReliableReceiver的方式保证不丢失数据，就是说在driver端打开chechpoint，用于定期的保存driver端的状态信息到HDFS上，保证driver端的状态信息不会丢失；在接收数据Receiver所在的Executor上打开WAL，使得接收到的数据保存在HDFS中，保证接收到的数据不会丢失；因为我们使用的是ReliableReceiver，所以在Receiver挂掉的期间，是不会接收数据，当这个Receiver重启的时候，会从上次消费的地方开始消费。

所以我们可以总结Spark Streaming的checkpoint机制包括driver端元数据的checkpoint以及Executor端的数据的checkpoint（WAL以及updateStateByKey等也需要checkpint）,Executor端的checkpoint机制除了保证数据写到HDFS之外，还有切断很长的RDD依赖的功效

Driver端checkpoint

待定：：

Executor端checkpoint

待定：：

Kafka Direct模式：

这种模式下，因为数据源都是存储在Kafka中的，所以一般不会丢数据，但是有一种情况下可能会丢失数据，就是当Spark Streaming应用失败后或者升级重启的时候因为没有记住重启之前消费的topic的offset，使得重启后Spark Streaming从topic的最新的offset开始消费(这个是默认的行为)，这样就导致Spark Streaming消费不到失败或者重启过程中Kafka接收到的消息，解决这个问题的办法有三个：

1、使用Spark Streaming自带的Driver端checkpoint机制，因为Driver端checkpoint机制会定期的保存Driver端的状态信息，当然也包括当前批次消费的Kafka中topic的offset信息啦，这样下次重启的时候就可以从checkpoint文件中直接读取上次消费到的offset信息，然后从这个offset开始消费。但是Driver端的checkpoint机制有一个很明显的缺陷，因为Driver端的checkpoint机制保存的Driver端的状态信息还包含DStreamGraph的状态信息，说白了就是将Driver端的代码序列化到checkpoint文件中，这样的话，如果我们对代码做了很大的改动或者升级的话，那么升级后的代码和checkpoint文件中的代码不兼容，这样的话会导致重启失败，解决这个问题的方法就是每次升级的时候将checkpoint文件清除掉，但是这样做的话也清除了保存在checkpoint文件中上次消费到的offset信息，这个不是我们想要的，所以这种方式不可取

2、我们可以在每一个批次开始之前将我们消费到的offset手动的保存到其他第三方存储系统中，可以是zookeeper或者Hbase，如下：