spark-streaming支援kafka消費，有以下方式：

我實驗的版本是kafka0.10，試驗的是spark-streaming-kafka-0.8的接入方式。另外，spark-streaming-kafka-0.10的分支並沒有研究。

spark-streaming-kafka-0.8的方式支援kafka0.8.2.1以及更高的版本。有兩種方式：
(1)Receiver Based Approach：基於kafka high-level consumer api，有一個Receiver負責接收資料到執行器
(2)Direct Approcah：基於kafka simple consumer api，沒有receiver。

mavne專案需要新增依賴

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

Reviced based approach程式碼：使用方法見註釋

package com.lgh.sparkstreaming

import org.apache
 
.spark.SparkConf
import org.apache.spark.streaming.{Minutes, Seconds, StreamingContext}
import org.apache.spark.streaming.kafka.KafkaUtils

/**
  * Created by Administrator on 2017/8/23.
  */
object KafkaWordCount {
  def main(args: Array[String]): Unit = {
    if (args.length < 4) {
      System.err.println
 
("Usage: KafkaWordCount <zkQuorum> <group> <topics> <numThreads>")
      System.exit(1)
    }
  //引數分別為 zk地址，消費者group名，topic名 多個的話，分隔 ，執行緒數
    val Array(zkQuorum, group, topics, numThreads) = args
    //setmaster，local是除錯模式使用
    val sparkConf = new SparkConf().setAppName("KafkaWordCount").setMaster("local[2]")
    val ssc = new StreamingContext(sparkConf, Seconds(2))
    ssc.checkpoint("checkpoint")

    //Map型別儲存的是   key： topic名字   values： 讀取該topic的消費者的分割槽數
    val topicMap = topics.split(",").map((_, numThreads.toInt)).toMap

    //引數分別為StreamingContext,kafka的zk地址，消費者group，Map型別
    val kafkamessage = KafkaUtils.createStream(ssc, zkQuorum, group, topicMap)
    //_._2取出kafka的實際訊息流
    val lines=kafkamessage.map(_._2)

    val words = lines.flatMap(_.split(" "))
    val wordCounts = words.map(x => (x, 1L))
      .reduceByKeyAndWindow(_ + _, _ - _, Minutes(10), Seconds(2), 2)
    wordCounts.print()

    ssc.start()
    ssc.awaitTermination()
  }
}

Direct approach：

package com.lgh.sparkstreaming

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

/**
  * Created by Administrator on 2017/8/23.
  */
object DirectKafkaWordCount {

    def main(args: Array[String]) {
      if (args.length < 2) {
        System.err.println(s"""
                              |Usage: DirectKafkaWordCount <brokers> <topics>
                              |  <brokers> is a list of one or more Kafka brokers
                              |  <topics> is a list of one or more kafka topics to consume from
                              |
        """.stripMargin)
        System.exit(1)
      }
     //borkers ： kafka的broker 列表,多個的話以逗號分隔
      //topics： kafka topic，多個的話以逗號分隔
      val Array(brokers, topics) = args

      // Create context with 2 second batch interval
      val sparkConf = new SparkConf().setAppName("DirectKafkaWordCount").setMaster("local[2]")
      val ssc = new StreamingContext(sparkConf, Seconds(2))

      // Create direct kafka stream with brokers and topics
      val topicsSet = topics.split(",").toSet
      val kafkaParams = Map[String, String]("metadata.broker.list" -> brokers)
      val messages = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
        ssc, kafkaParams, topicsSet)

      // Get the lines, split them into words, count the words and print
      val lines = messages.map(_._2)
      val words = lines.flatMap(_.split(" "))
      val wordCounts = words.map(x => (x, 1L)).reduceByKey(_ + _)
      wordCounts.print()

      // Start the computation
      ssc.start()
      ssc.awaitTermination()

  }

}

關於這兩種方式的區別

1.Simplified Parallelism
Direct 方式將會建立跟kafka分割槽一樣多的RDD partiions，並行的讀取kafka topic的partition資料。kafka和RDD partition將會有一對一的對應關係。
2.Efficiency
Receiver-based Approach需要啟用WAL才能保證消費不丟失資料
，效率比較低
3.Exactly-once semantics
Receiver-based Approach使用kafka high-level consumer api，儲存消費者offset在zookeeper中，跟Write Ahead Log配合使用，能夠實現至少消費一次語義。
Direct Approach 使用kafka simple consumer api，跟蹤offset資訊儲存在spark checkpoint中。能夠實現資料有且只消費一次語義。