方式一  Receiver
          使用Kafka的高層次Consumer API來實現。receiver從Kafka中獲取的資料都儲存在Spark Executor的記憶體中，然後Spark Streaming啟動的job會去處理那些資料。然而，在預設的配置下，這種方式可能會因為底層的失敗而丟失資料。如果要啟用高可靠機制，讓資料零丟失，就必須啟用Spark Streaming的預寫日誌機制（Write Ahead Log，WAL）。該機制會同步地將接收到的Kafka資料寫入分散式檔案系統（比如HDFS）上的預寫日誌中。所以，即使底層節點出現了失敗，也可以使用預寫日誌中的資料進行恢復。

注意事項：

1、Kafka中topic的partition與Spark中RDD的partition是沒有關係的，因此，在KafkaUtils.createStream()中，提高partition的數量，只會增加Receiver的數量，也就是讀取Kafka中topic partition的執行緒數量，不會增加Spark處理資料的並行度。

2、可以建立多個Kafka輸入DStream，使用不同的consumer group和topic，來通過多個receiver並行接收資料。

3、如果基於容錯的檔案系統，比如HDFS，啟用了預寫日誌機制，接收到的資料都會被複制一份到預寫日誌中。因此，在KafkaUtils.createStream()中，設定的持久化級別是StorageLevel.MEMORY_AND_DISK_SER。

程式碼展示：

val conf = new SparkConf().setAppName("kafkawc").setMaster("local[2]")
val ssc = new StreamingContext(conf,Duration(5000))
ssc.checkpoint("xxx")
//編寫kafka配置資訊
//首先我們編寫kafka依賴的zk資訊
val zks = "192.168.244.141:2181"  //3個全部寫上也行
//然後是kafka的消費者組
val groupId = "gp1"
  //Topic名字
val topics = Map[String,Int]("demo1"->1)
//建立kafka的輸入資料流，獲取kafka的資料  第一種方式讀取kafka直接createStream
val data: ReceiverInputDStream[(String, String)] = KafkaUtils.createStream(ssc,zks,groupId,topics)

方式二  Direct

          Spark1.3中引入Direct方式，用來替代掉使用Receiver接收資料，這種方式會週期性地查詢Kafka，獲得每個topic+partition的最新的offset，從而定義每個batch的offset的範圍。當處理資料的job啟動時，就會使用Kafka的簡單consumer api來獲取Kafka指定offset範圍的資料。

優點

1、簡化並行讀取：如果要讀取多個partition，不需要建立多個輸入DStream，然後對它們進行union操作。Spark會建立跟Kafka partition一樣多的RDD partition，並且會並行從Kafka中讀取資料。所以在Kafka partition和RDD partition之間，有一個一對一的對映關係。

2、高效能：如果要保證零資料丟失，在基於receiver的方式中，需要開啟WAL機制。這種方式其實效率低下，因為資料實際上被複制了兩份，Kafka自己本身就有高可靠的機制會對資料複製一份，而這裡又會複製一份到WAL中。而基於direct的方式，不依賴Receiver，不需要開啟WAL機制，只要Kafka中作了資料的複製，那麼就可以通過Kafka的副本進行恢復。

程式碼：

 Logger.getLogger("org").setLevel(Level.WARN)
    val conf = new SparkConf()
    conf.setAppName("spark_streaming")
    conf.setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setCheckpointDir("e:/checkDir")
    sc.setLogLevel("ERROR")
    val ssc = new StreamingContext(sc, Seconds(5))
    // val topics = Map("spark" -> 2)
    val kafkaParams = Map[String, String](
      "bootstrap.servers" -> "192.168.244.141:9092",
      "group.id" -> "wc",
      "auto.offset.reset" -> "smallest"
    )
    // 直連方式拉取資料，這種方式不會修改資料的偏移量，需要手動的更新  //第一個值是偏移量 第二個是資料
    //第一個引數是streamingContext，第二個是叢集一些配置，第三個是topic主題，可以有多個
    val lines: DStream[String] =  KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, Set("wc2")).map(_._2)
    // val lines = KafkaUtils.createStream(ssc, "m1:2181,m2:2181,m3:2181", "spark", topics).map(_._2)
//下面根據lines進行操作lines的型別是DStream[String]
    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()