有提到过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分副本。