Spark-Streaming获取kafka数据的两种方式：Receiver与Direct的方式

 简单理解为：Receiver方式是经过zookeeper来链接kafka队列，Direct方式是直接链接到kafka的节点上获取数据

Receiver

使用Kafka的高层次Consumer API来实现。receiver从Kafka中获取的数据都存储在Spark Executor的内存中，而后Spark Streaming启动的job会去处理那些数据。然而，在默认的配置下，这种方式可能会由于底层的失败而丢失数据。若是要启用高可靠机制，让数据零丢失，就必须启用Spark Streaming的预写日志机制（Write Ahead Log，WAL）。该机制会同步地将接收到的Kafka数据写入分布式文件系统（好比HDFS）上的预写日志中。因此，即便底层节点出现了失败，也可使用预写日志中的数据进行恢复。

注意事项：

一、Kafka中topic的partition与Spark中RDD的partition是没有关系的，所以，在KafkaUtils.createStream()中，提升partition的数量，只会增长Receiver的数量，也就是读取Kafka中topic partition的线程数量，不会增长Spark处理数据的并行度。

二、能够建立多个Kafka输入DStream，使用不一样的consumer group和topic，来经过多个receiver并行接收数据。

三、若是基于容错的文件系统，好比HDFS，启用了预写日志机制，接收到的数据都会被复制一份到预写日志中。所以，在KafkaUtils.createStream()中，设置的持久化级别是StorageLevel.MEMORY_AND_DISK_SER。

Direct

Spark1.3中引入Direct方式，用来替代掉使用Receiver接收数据，这种方式会周期性地查询Kafka，得到每一个topic+partition的最新的offset，从而定义每一个batch的offset的范围。当处理数据的job启动时，就会使用Kafka的简单consumer api来获取Kafka指定offset范围的数据。

这种方式有以下优势：

一、简化并行读取：若是要读取多个partition，不须要建立多个输入DStream，而后对它们进行union操做。Spark会建立跟Kafka partition同样多的RDD partition，而且会并行从Kafka中读取数据。因此在Kafka partition和RDD partition之间，有一个一对一的映射关系。

二、高性能：若是要保证零数据丢失，在基于receiver的方式中，须要开启WAL机制。这种方式其实效率低下，由于数据实际上被复制了两份，Kafka本身自己就有高可靠的机制会对数据复制一份，而这里又会复制一份到WAL中。而基于direct的方式，不依赖Receiver，不须要开启WAL机制，只要Kafka中做了数据的复制，那么就能够经过Kafka的副本进行恢复。

三、一次且仅一次的事务机制：基于receiver的方式，是使用Kafka的高阶API来在ZooKeeper中保存消费过的offset的。这是消费Kafka数据的传统方式。这种方式配合着WAL机制能够保证数据零丢失的高可靠性，可是却没法保证数据被处理一次且仅一次，可能会处理两次。由于Spark和ZooKeeper之间多是不一样步的。基于direct的方式，使用kafka的简单api，Spark Streaming本身就负责追踪消费的offset，并保存在checkpoint中。Spark本身必定是同步的，所以能够保证数据是消费一次且仅消费一次。因为数据消费偏移量是保存在checkpoint中，所以，若是后续想使用kafka高级API消费数据，须要手动的更新zookeeper中的偏移量

Direct代码

package bigdata.spark

import kafka.serializer.{StringDecoder, Decoder}
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkContext, SparkConf}

import scala.reflect.ClassTag

/**
  * Created by Administrator on 2017/4/28.
  */
object SparkStreamDemo {
  def main(args: Array[String]) {

    val conf = new SparkConf()
    conf.setAppName("spark_streaming")
    conf.setMaster("local[*]")

    val sc = new SparkContext(conf)
    sc.setCheckpointDir("D:/checkpoints")
    sc.setLogLevel("ERROR")

    val ssc = new StreamingContext(sc, Seconds(5))

    // val topics = Map("spark" -> 2)

    val kafkaParams = Map[String, String](
      "bootstrap.servers" -> "m1:9092,m2:9092,m3:9092",
      "group.id" -> "spark",
      "auto.offset.reset" -> "smallest"
    )
    // 直连方式拉取数据，这种方式不会修改数据的偏移量，须要手动的更新
    val lines =  KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, Set("spark")).map(_._2)
    // val lines = KafkaUtils.createStream(ssc, "m1:2181,m2:2181,m3:2181", "spark", topics).map(_._2)

    val ds1 = lines.flatMap(_.split(" ")).map((_, 1))

    val ds2 = ds1.updateStateByKey[Int]((x:Seq[Int], y:Option[Int]) => {
      Some(x.sum + y.getOrElse(0))
    })

    ds2.print()

    ssc.start()
    ssc.awaitTermination()

  }
}