第4课：Spark Streaming的Exactly-One的事务处理和不重复输出彻底掌握

[b] 作者：大数据研究人员：谢彪
[/b]

一、事务场景 :

　　以银行转帐一次为例，A用户转账给B用户，如何保证事务的一致性，即A用户能够转出且只能转出一次，B用户能够收到且只能收到一次。

   SparkStreaming的事务处理和关系型数据库的事务的概念有所不同，关系型数据库事务关注的是语句级别的一致性，例如银行转账。而SparkStreaming的事务关注的是某次job执行的一致性。​

SparkStreaming运行架构流程图：


运行架构流程图

ExactlyOnce的事务处理：

    1、数据零丢失：必须有可靠的数据来源和可靠的Receiver，且整个应用程序的metadata必须进行checkpoint，且通过WAL来保证数据安全。

其中Checkpoint的属性：

private[streaming]class Checkpoint(ssc: StreamingContext, val checkpointTime: Time)  extends Logging with Serializable {  val master = ssc.sc.master  val framework = ssc.sc.appName
  val jars = ssc.sc.jars​

 val checkpointDir = ssc.checkpointDir

  val checkpointDuration = ssc.checkpointDuration  val pendingTimes = ssc.scheduler.getPendingTimes().toArray  val delaySeconds = MetadataCleaner.getDelaySeconds(ssc.conf)
  val sparkConfPairs = ssc.conf.getAll​

其中graph是DStreamGraph的实例化，它里面包含了InputDStream

private val inputStreams = new ArrayBuffer[InputDStream[_]]()
我们以DirectKafkaInputDStream为例，其中包含了checkpointData

protected[streaming] override val checkpointData =  new DirectKafkaInputDStreamCheckpointData
其中只是包含：​

class DirectKafkaInputDStreamCheckpointData extends DStreamCheckpointData(this) {

  def batchForTime: mutable.HashMap[Time, Array[(String, Int, Long, Long)]] = {    data.asInstanceOf[mutable.HashMap[Time, Array[OffsetRange.OffsetRangeTuple]]]

checkpoint是非常高效的。没有涉及到实际数据的存储。一般大小只有几十K，因为只存了Kafka的偏移量等信息。​

 checkpoint 采用的是序列化机制，尤其是DStreamGraph的引入，里面包含了可能如ForeachRDD等，而ForeachRDD里面的函数应该也会被序列化。如果采用了CheckPoint机制，而你的程序包做了做了变更，恢复后可能会有一定的问

2、Spark Streaming1.3的时候为了避免WAL的性能损失和实现ExactlyOnce而提供了Kafka DirectAPI，把Kafka作为文件存储系统！！！此时兼具有流的优势和文件系统的优势，至此，SparkStreaming+Kafka就构建了完美的流处理世界！！！所有的Executors通过KafkaAPI直接消费数据，直接管理Offset，所以也不会重复消费数据；实现实时务事啦！！！

二、如何保证Job在处理数据的过程中保证：

    一是不丢失数据，二是不重复处理数据

    1、数据丢失及其具体的解决方式：在Receiver收到数据且通过Driver的调度Executor开始计算数据的时候如果Driver突然崩溃，则此时Executor会被Kill掉，那么Executor中的数据就会丢失，此时就必须通过例如WAL的方式让所有的数据都通过例如HDFS的方式首先进行安全性容错处理，此时如果Executor中的数据丢失的话就可以通过WAL恢复回来。​

 2、数据重复读取的情况：在Receiver收到数据且保存到了HDFS等持久化引擎但是没有来得及进行updateOffsets，此时Receiver崩溃后重新启动就会通过管理Kafka的ZooKeeper中元数据再次重复读取数据，但是此时SparkStreaming认为是成功的，但是Kafka认为是失败的（因为没有更新offset到ZooKeeper中），此时就会导致数据重新消费的情况。

   三、性能损失：

   1、通过WAL方式会极大的损伤SparkStreaming中Receivers接受数据的性能；

   2、如果通过Kafka的作为数据来源的话，Kafka中有数据，然后Receiver接受的时候又会有数据副本，这个时候其实是存储资源的浪费。​

 四、关于SparkStreaming数据输出多次重写及其解决方案：

    1、为什么会有这个问题，因为SparkStreaming在计算的时候基于SparkCore，SparkCore天生会做以下事情导致SparkStreaming的结果（部分）重复输出：

    Task重试；

    慢任务推测

    Stage重复；

    Job重试；

  2、具体解决方案：

    设置spark.task.maxFailures次数为1；

    设置spark.speculation为关闭状态（因为慢任务推测其实非常消耗性能，所以关闭后可以显著提高SparkStreaming处理性能）

    SparkStreaming onKafka的话，Job失败后可以设置auto.offset.reset为“largest”的方式；

    最后再次强调可以通过transform和foreachRDD基于业务逻辑代码进行逻辑控制来实现数据不重复消费和输出不重复！这两个方式类似于SparkStreaming的后门，可以做任意想象的控制操作！

 作者：大数据研究人员：谢彪

资料来源于：DT_大数据梦工厂(Spark发行版本定制) 

DT大数据梦工厂微信公众号:DT_Spark 

新浪微博：http://www.weibo.com/ilovepains

王家林老师每晚20:00免费大数据实战​

YY直播：68917580
99f5