需要导入pom依赖

<dependency>
     <groupId>org.apache.spark</groupId>
     <artifactId>spark-streaming-kafka-0-10_2.11</artifactId>
     <version>2.0.2</version>
</dependency>

一、receiver方式读取

Receiver获取数据，是很早之前的老方法。原理上使用Kafka的高层次 API来实现数据的消费，也就是从Kafka中获取的数据存储在Spark Executor的内存中的，然后Spark Streaming启动的job会去处理那些数据，和kafka的关联力度不强

而且有一个弊端，这种方式可能会因为job的失败而丢失数据，因为计算进行中kafka的高级API拉取到的数据偏移量是不会在第一时间被更新到kafka的，是周期性的更新

因此要启用高可靠机制，让数据的消费出现问题是可以从高可用中恢复，比如启用Spark Streaming的预写日志机制（Write Ahead Log，WAL）。该机制会同步地将接收到的Kafka数据写入分布式文件系统（比如HDFS）上的预写日志中。所以，即使spark任务出现了失败，也可以使用预写日志中的数据进行失败计算的恢复，而不需要再次从kafka拉取，那还要考虑偏移量

而对于使用这种方式需要注意的要点如下

1、这种方式下Kafka中被消费topic的partition与Spark中的RDD的partition是没有直接关系的。所以在KafkaUtils.createStream()的参数中，提高的partition的数量，只是增加Receiver读取kafka_partition的线程数量不会增加消费者进程

2、想要增加消费者可以创建多个Kafka输入DStream，使用相同的consumer group和topic，来通过人为增加的多个receiver并行接收数据。

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

4、在比较新的版本jar中KafkaUtils已经将receiver方式读取的api删除了

虽然被删除了，但是我们还是要知道实现方式：

package com.spark

import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkContext, SparkConf}


object SparkStreamingReceiverKafka {
  def main(args: Array[String]) {
    val conf = new SparkConf()
    conf.setAppName("SparkStreamingReceiverKafka")
    conf.set("spark.streaming.kafka.maxRatePerPartition", "10")
	conf.set("spark.streaming.receiver.writeAheadLog.enable", "true")//预写日志需要改这个配置
    conf.setMaster("local[2]")

    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

    val ssc = new StreamingContext(sc, Seconds(5)) // 创建streamingcontext入口

	ssc.checkpoint("hdfs://localhost:9000/log")//预写日志的hdfs地址需要通过checkpoint设置
	
    val zks = "zk1,zk2,zk3"
    val groupId = "kafka_spark_xf"
    val map : Map[String, Int] = Map("kafka_spark" -> 2) // topic名称为kafka_spark，每次使用2个线程读取数据

	//参数： 流对象 zookeeper集群 消费者id map参数，日志等级必须要有落盘操作
    val dframe = KafkaUtils.createStream(ssc, zks, groupId, map, StorageLevel.MEMORY_AND_DISK_SER_2)
    
    dframe.foreachRDD(rdd => { // 操作方式和rdd差别不大
      rdd.foreachPartition(partition =>{
        partition.foreach(println)
      })
    })
  }
}

二、direct方式读取

这种新的不基于Receiver的方式，是在Spark 1.3中引入的，这种方式在当前也叫做直连，或者是低级 API，它能够确保更加健壮的机制。在这种方式下spark默认随着消费数据的任务，紧随其后的对Kafka的offset进行同步维护，不在是周期性的。通俗理解为消费一个窗口数据就提交一次偏移量。

这种方式有如下优点：

1、简化并行读取，如果要读取多个partition，不需要创建多个输入DStream然后对它们进行union操作。Spark默认会创建跟Kafka partition一样多的RDD partition，并且会并行从Kafka中读取数据。所以在Kafka partition和RDD partition之间，有一个一对一的映射关系。

2、高性能，保证零数据丢失，在基于receiver的方式中，需要开启WAL机制。这种方式效率低下，因为数据实际上被复制了两份，Kafka自己本身就有高可靠的机制，会对数据复制一份，而这里又会复制一份到WAL中 ，这对效率的消耗是致命的。而direct的方式，不需要开启WAL机制，依靠Kafka本身有数据高可用的同时，即使任务在运行中出现了kafka到spark之间的数据意外，那么也会因为偏移量是随着消费数据后同步更新的，以及我们可以做其他辅助标识，而直观的知道在哪个窗口的数据丢了，通过Kafka的副本或者指定偏移量拉取进行恢复。不过通常也不会很麻烦，因为一般一旦出现意外Spark任务就失败了，偏移量也不会出现空缺，调整代码重启任务就行。

3、一次且仅一次的事务机制，基于receiver的方式，是使用Kafka的高阶API来在ZooKeeper中保存消费过的offset的。这是消费Kafka数据的老旧方式。这种方式配合着WAL机制可以保证数据零丢失的高可靠性，但是却无法保证数据被处理一次且仅一次。因为receiver提交offset是一个周期性的并不及时，Spark和ZooKeeper之间可能由于某种原因导致spark还没有更新而导致偏移量不同步，就是说可能存在zk上offset是0 1024 ，spark同步的时候自身可能已经消费到0 1500 ，那么没有同步的1023 1500就会发生很多意外。
而基于direct的方式，使用kafka的简单api，Spark Streaming自己就负责追踪消费的offset，并默认时保存在内存中，并且Spark随着消费任务进行和kafka同步，因此可以保证数据是消费一次且仅消费一次，这也就是牵扯到我们常说的自己维护偏移量，这里也是一个面试的点，面试官问你为什么它可以保证数据不被重复消费，其实就是因为运行时，spark自身就随着任务的数据消费维护有一份相对准确的偏移量，保障着数据消费且只消费一次，这也是为什么我们现在要手动维护偏移量，我们维护的目的就是把spark自己知道的这份相对准确的偏移量维护到其他地方为开发在上一道保险，而且要养成写代码的时候习惯性的设置一个检查点路径，不是把它放在默认的内存中

package com.stream

import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe


object StreamFromKafka {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("StreamWordCount").setMaster("local[2]")
    val sc = new StreamingContext(conf,Seconds(10))

	sc.checkpoint("hdfs://localhost:9000/log")

    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "192.168.182.146:9092,192.168.182.147:9092,192.168.182.148:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "group1"
    )

    /**
      * LocationStrategies.PreferBrokers() 是当在你kafka集群和spark集群部署节点存在同一服务器上的时候，任务的 executor 会优先运行该节点上，节省了网络IO，缺点是容易造成资源热点问题，不推荐使用。
      * LocationStrategies.PreferConsistent() 大多数情况下使用，不考虑kafka集群所在的影响，均匀的分配分区所有 executor 在spark集群上。
      * 新的Kafka使用者API将预先获取消息到缓冲区。因此，出于性能原因，Spark集成将缓存的消费者保留在执行程序上（而不是为每个批处理重新创建它们），并且更喜欢在具有适当使用者的主机位置上安排分区，这一点很重要。
      *在大多数情况下，您应该使用LocationStrategies.PreferConsistent，如上所示。这将在可用执行程序之间均匀分配分区。如果您的执行程序与Kafka代理在同一主机上，请使用PreferBrokers，它更愿意为该分区安排Kafka领导者的分区。
      */
    val topics = Array("test")
    val stream = KafkaUtils.createDirectStream[String, String](
      sc,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )
    val kafkaStream = stream.map(record => (record.key, record.value))
    val words = kafkaStream.map(_._2)
    val pairs = words.map { x => (x,1) }
    val wordCounts = pairs.reduceByKey(_+_)
    wordCounts.print()
    sc.start()
    sc.awaitTermination()
  }

}