Spark Streaming源码解读之No Receivers

前文有提到过Spark Streaming事务是如何保证exactly once的语义的。

从spark core程序来说，读取固定数据来源好比hdfs中，spark只是作为一个计算框架。

而在流处理中，只是多了一个时间维度。

若在某一时刻，知道所需处理数据的来源，直接读取，而不用被动的接收（Receiver），那就是和普通的Spark 程序没什么差异了。

本文将着重Kafka中direct方式的读取，以案例切入，跟踪源码分析。

入口是KafkaUtils，先建立了一个回调函数定义，再获取到kafka集群，并获取到起始偏移量，最后建立一个DirectKafkaInputDStream，用于建立RDD。

// KafkaUtils.scala line 473
  def createDirectStream[
    K: ClassTag,
    V: ClassTag,
    KD <: Decoder[K]: ClassTag,
    VD <: Decoder[V]: ClassTag] (
      ssc: StreamingContext,
      kafkaParams: Map[String, String],
      topics: Set[String]
  ): InputDStream[(K, V)] = {
    val messageHandler = (mmd: MessageAndMetadata[K, V]) => (mmd.key, mmd.message)
    val kc = new KafkaCluster(kafkaParams)
    val fromOffsets = getFromOffsets(kc, kafkaParams, topics)
    new DirectKafkaInputDStream[K, V, KD, VD, (K, V)](
      ssc, kafkaParams, fromOffsets, messageHandler)
  }

KafkaCluster 在实例化时没有任何动做，只是单纯的建立对象。

接下来获取每一个partition的偏移量，

1. 先获取设置的偏移量配置
2. 获取分区，使用的是Either，即要么left要么right。此处left为抛出一个异常，right为返回TopicAndPartition Set。scala语法博大精深。，，后面的大量用到了Either。业务逻辑没什么复杂。
3. 偏移量默认是最大的。

DStream建立以后，整个DAG回溯的lineage以下：

DirectKafkaInputDStream -> MappedDStream > FlatMappedDStream -> MappedDStream -> ShuffledDStream -> ForEachDStream

当DAG回溯到DirectKafkaInputDStream时，会调用compute。建立KafkaRDD，而且将最新的偏移量保存，以便下次计算新的偏移量。

从当RDD进入计算时，会调用compute。，此处的offsetRanges就是Kafka的TopicAndPartition和对应的偏移量。最后结果就是kafka有多少个partition，spark就会有多少个partition与之对应。

 

直接抓Kafka数据的方式与Receiver的方式的对比：

1. 实现方式
   1. Direct方式只是定时生产RDD，经过RDD 回溯至最先的KafkaRDD来获取数据，是很天然的作法
   2. Receiver是以RDD的形式封装Receiver，在Worker中启动后，未到时的数据存在内存队列中，定时触发来收割数据，放入Spark中
2. 数据可靠性
   1. Direct数据存放在外部Kafka中，kafka自带副本。无需spark作冗余
   2. Receiver须要WAL来预防数据丢失，可是由于wal的批处理的特性，仍是有可能丢失数据。
3. 负载均衡
   1. Direct以RDD的形式，每一个Duration产生的RDD都会在执行时动态最优。
   2. Receiver 与 Worker绑定，没法动态调整。
4. 一致性保证
   1. direct借助kafka的ack，失败的消息会自动重试。
   2. 借助wal实现一致性。
5. 对资源的合理使用
   1. direct方式数据都存在kafka，没冗余，没wal，
   2. receiver未收割的数据存在内存queue中，必要时要开启wal，至少多了1分副本。