论Spark Streaming的数据可靠性和一致性

摘要：Spark Streaming自发布起就获得了普遍的关注，然而做为一个年轻的项目，须要提高的地方一样不少，好比1.2以前版本driver挂掉可能会丢失数据。这里将分析它的可靠性机制。缓存

眼下大数据领域最热门的词汇之一即是流计算了，其中最耀眼的项目无疑是来自Spark社区的Spark Streaming项目，其从一诞生就受到普遍关注并迅速发展，目前已有追赶并超越Storm的架势。安全

对于流计算而言，毫无疑问最核心的特色是它的低时延能力，这主要是来自对数据不落磁盘就进行计算的内部机制，但这也带来了数据可靠性的问题，即有节点失效或者网络异常时，如何在节点间进行合适的协商来进行重传。更进一步的，若发生计划外的数据重传，怎么能保证没有产生重复的数据，全部数据都是精确一次的（Exact Once）？若是不解决这些问题，大数据的流计算将没法知足大多数企业级可靠性要求而流于徒有虚名。微信

本文将重点分析Spark Streaming是如何设计可靠性机制并实现数据一致性的。网络

Driver HA

因为流计算系统是长期运行、数据不断流入的，所以其Spark守护进程（Driver）的可靠性是相当重要的，它决定了Streaming程序可否一直正确地运行下去。运维

图一 Driver数据持久化机器学习

Driver实现HA的解决方案就是将元数据持久化，以便重启后的状态恢复。如图一所示，Driver持久化的元数据包括：socket

Block元数据（图一中的绿色箭头）：Receiver从网络上接收到的数据，组装成Block后产生的Block元数据；编辑器
Checkpoint数据（图一中的橙色箭头）：包括配置项、DStream操做、未完成的Batch状态、和生成的RDD数据等；ide

图二 Driver故障恢复oop

Driver失败重启后：

恢复计算（图二中的橙色箭头）：使用Checkpoint数据重启driver，从新构造上下文并重启接收器。
恢复元数据块（图二中的绿色箭头）：恢复Block元数据。
恢复未完成的做业（图二中的红色箭头）：使用恢复出来的元数据，再次产生RDD和对应的job，而后提交到Spark集群执行。

经过如上的数据备份和恢复机制，Driver实现了故障后重启、依然能恢复Streaming任务而不丢失数据，所以提供了系统级的数据高可靠。

可靠的上下游IO系统

流计算主要经过网络socket通讯来实现与外部IO系统的数据交互。因为网络通讯的不可靠特色，发送端与接收端须要经过必定的协议来保证数据包的接收确认、和失败重发机制。

不是全部的IO系统都支持重发，这至少须要实现数据流的持久化，同时还要实现高吞吐和低时延。在Spark Streaming官方支持的data source里面，能同时知足这些要求的只有Kafka，所以在最近的Spark Streaming release里面，也是把Kafka当成推荐的外部数据系统。

除了把Kafka当成输入数据源（inbound data source）以外，一般也将其做为输出数据源（outbound data source）。全部的实时系统都经过Kafka这个MQ来作数据的订阅和分发，从而实现流数据生产者和消费者的解耦。

一个典型的企业大数据中心数据流向视图以下所示：

图三企业大数据中心数据流向视图

除了从源头保证数据可重发以外，Kafka更是流数据Exact Once语义的重要保障。Kafka提供了一套低级API，使得client能够访问topic数据流的同时也能访问其元数据。Spark Streaming的每一个接收任务能够从指定的Kafka topic、partition和offset去获取数据流，各个任务的数据边界很清晰，任务失败后能够从新去接收这部分数据而不会产生“重叠的”数据，于是保证了流数据“有且仅处理一次”。

可靠的接收器

在Spark 1.3版本以前，Spark Streaming是经过启动专用的Receiver任务来完成从Kafka集群的数据流拉取。

Receiver任务启动后，会使用Kafka的高级API来建立topicMessageStreams对象，并逐条读取数据流缓存，每一个batchInerval时刻到来时由JobGenerator提交生成一个spark计算任务。

因为Receiver任务存在宕机风险，所以Spark提供了一个高级的可靠接收器-ReliableKafkaReceiver类型来实现可靠的数据收取，它利用了Spark 1.2提供的WAL（Write Ahead Log）功能，把接收到的每一批数据持久化到磁盘后，更新topic-partition的offset信息，再去接收下一批Kafka数据。万一Receiver失败，重启后还能从WAL里面恢复出已接收的数据，从而避免了Receiver节点宕机形成的数据丢失（如下代码删除了细枝末节的逻辑）：

class ReliableKafkaReceiver{  private var topicPartitionOffsetMap: mutable.HashMap[TopicAndPartition, Long] = null  private var blockOffsetMap: ConcurrentHashMap[StreamBlockId, Map[TopicAndPartition, Long]] = null  override def onStart(): Unit = {    // Initialize the topic-partition / offset hash map.    topicPartitionOffsetMap = new mutable.HashMap[TopicAndPartition, Long]    // Initialize the block generator for storing Kafka message.    blockGenerator = new BlockGenerator(new GeneratedBlockHandler, streamId, conf)    messageHandlerThreadPool = Utils.newDaemonFixedThreadPool(      topics.values.sum, "KafkaMessageHandler")    blockGenerator.start()    val topicMessageStreams = consumerConnector.createMessageStreams(      topics, keyDecoder, valueDecoder)    topicMessageStreams.values.foreach { streams =>      streams.foreach { stream =>        messageHandlerThreadPool.submit(new MessageHandler(stream))      }    }  }

启用WAL后虽然Receiver的数据可靠性风险下降了，但却因为磁盘持久化带来的开销，系统总体吞吐率会有明显的降低。所以，在最新发布的Spark 1.3版本里，Spark Streaming增长了使用Direct API的方式来实现Kafka数据源的访问。

引入了Direct API后，Spark Streaming再也不启动常驻的Receiver接收任务，而是直接分配给每一个Batch及RDD最新的topic partition offset。job启动运行后Executor使用Kafka的simple consumer API去获取那一段offset的数据。

这样作的好处不只避免了Receiver宕机带来的数据可靠性风险，同时也因为避免使用ZooKeeper作offset跟踪，而实现了数据的精确一次性（如下代码删除了细枝末节的逻辑）：

class DirectKafkaInputDStream{  protected val kc = new KafkaCluster(kafkaParams)  protected var currentOffsets = fromOffsets  override def compute(validTime: Time): Option[KafkaRDD[K, V, U, T, R]] = {    val untilOffsets = clamp(latestLeaderOffsets(maxRetries))    val rdd = KafkaRDD[K, V, U, T, R](      context.sparkContext, kafkaParams, currentOffsets, untilOffsets, messageHandler)    currentOffsets = untilOffsets.map(kv => kv._1 -> kv._2.offset)    Some(rdd)  }

预写日志 Write Ahead Log

Spark 1.2开始提供了预写日志能力，用于Receiver数据及Driver元数据的持久化和故障恢复。WAL之因此能提供持久化能力，是由于它利用了可靠的HDFS作数据存储。

Spark Streaming预写日志机制的核心API包括：

管理WAL文件的WriteAheadLogManager
读/写WAL的WriteAheadLogWriter和WriteAheadLogReader
基于WAL的RDD：WriteAheadLogBackedBlockRDD
基于WAL的Partition：WriteAheadLogBackedBlockRDDPartition

以上核心API在数据接收和恢复阶段的交互示意图如图四所示。

图四基于WAL的数据接收和恢复示意图

从WriteAheadLogWriter的源码里能够清楚地看到，每次写入一块数据buffer到HDFS后都会调用flush方法去强制刷入磁盘，而后才去取下一块数据。所以receiver接收的数据是能够保证持久化到磁盘了，于是作到了较好的数据可靠性。

private[streaming] class WriteAheadLogWriter{  private lazy val stream = HdfsUtils.getOutputStream(path, hadoopConf)  def write(data: ByteBuffer): WriteAheadLogFileSegment = synchronized {    data.rewind() // Rewind to ensure all data in the buffer is retrieved    val lengthToWrite = data.remaining()    val segment = new WriteAheadLogFileSegment(path, nextOffset, lengthToWrite)    stream.writeInt(lengthToWrite)    if (data.hasArray) {      stream.write(data.array())    } else {      while (data.hasRemaining) {        val array = new Array[Byte](data.remaining)        data.get(array)        stream.write(array)      }    }    flush()    nextOffset = stream.getPos()    segment  }

结束语

得益于Kafka这类可靠的data source、以及自身的checkpoint/WAL等机制，Spark Streaming的数据可靠性获得了很好的保证，数据能保证“至少一次”（at least once）被处理。但因为其outbound端的一致性实现还未完善，所以Exact once语义仍然不能端到端保证。Spark Streaming社区已经在跟进这个特性的实现（SPARK-4122），预计很快将合入trunk发布。