有興趣想學習國內整套Spark+Spark Streaming+Machine learning最頂級課程的，可加我qq  471186150。共享視訊，價效比超高！

1：Direct方式特點：

1）Direct的方式是會直接操作kafka底層的元資料資訊，這樣如果計算失敗了，可以把資料重新讀一下，重新處理。即資料一定會被處理。拉資料，是RDD在執行的時候直接去拉資料。

2）由於直接操作的是kafka，kafka就相當於你底層的檔案系統。這個時候能保證嚴格的事務一致性，即一定會被處理，而且只會被處理一次。而Receiver的方式則不能保證，因為Receiver和ZK中的資料可能不同步，spark Streaming可能會重複消費資料，這個調優可以解決，但顯然沒有Direct方便。而Direct api直接是操作kafka的，spark streaming自己負責追蹤消費這個資料的偏移量或者offset，並且自己儲存到checkpoint，所以它的資料一定是同步的，一定不會被重複。即使重啟也不會重複，因為checkpoint了，但是程序升級的時候，不能讀取原先的checkpoint，面對升級checkpoint無效這個問題，怎麼解決呢?升級的時候讀取我指定的備份就可以了，即手動的指定checkpoint也是可以的，這就再次完美的確保了事務性，有且僅有一次的事務機制。那麼怎麼手動checkpoint呢？構建SparkStreaming的時候，有getorCreate這個api，它就會獲取checkpoint的內容，具體指定下這個checkpoint在哪就好了。或者如下圖：

而如果從checkpoint恢復後，如果資料累積太多處理不過來，怎麼辦?1）限速2）增強機器的處理能力3）放到資料緩衝池中。

3）由於底層是直接讀資料，沒有所謂的Receiver，直接是週期性(Batch Intervel)的查詢kafka，處理資料的時候，我們會使用基於kafka原生的Consumer  api來獲取kafka中特定範圍(offset範圍)中的資料。這個時候，Direct Api訪問kafka帶來的一個顯而易見的效能上的好處就是，如果你要讀取多個partition，Spark也會建立RDD的partition，這個時候RDD的partition和kafka的partition是一致的。而Receiver的方式，這2個partition是沒任何關係的。這個優勢是你的RDD，其實本質上講在底層讀取kafka的時候，kafka的partition就相當於原先hdfs上的一個block。這就符合了資料本地性。RDD和kafka資料都在這邊。所以讀資料的地方，處理資料的地方和驅動資料處理的程式都在同樣的機器上，這樣就可以極大的提高效能。不足之處是由於RDD和kafka的patition是一對一的，想提高並行度就會比較麻煩。提高並行度還是repartition，即重新分割槽，因為產生shuffle，很耗時。這個問題，以後也許新版本可以自由配置比例，不是一對一。因為提高並行度，可以更好的利用叢集的計算資源，這是很有意義的。

4）不需要開啟wal機制，從資料零丟失的角度來看，極大的提升了效率，還至少能節省一倍的磁碟空間。從kafka獲取資料，比從hdfs獲取資料，因為zero copy的方式，速度肯定更快。

2：實戰部分

kafka + spark streaming 叢集

前提：

spark 安裝成功，spark 1.6.0

zookeeper 安裝成功 

kafka 安裝成功

步驟：

1：先啟動三臺機器上的ZK，然後三臺機器同樣啟動kafka，

2：在kafka上建立topic  test

3：在worker1中啟動kafka 生產者：

[email protected]:/usr/local/kafka_2.10-0.9.0.1# bin/kafka-console-producer.sh --broker-list localhost:9092

在worker2中啟動消費者：

[email protected]:/usr/local/kafka_2.10-0.9.0.1# bin/kafka-console-consumer.sh --zookeeper master:2181 --topic test

生產者生產的訊息，消費者可以消費到。說明kafka叢集沒問題。進入下一步。

在master中啟動spark-shell

./spark-shell --master local[2] --packages org.apache.spark:spark-streaming-kafka_2.10:1.6.0,org.apache.kafka:kafka_2.10:0.8.2.1

筆者用的spark 是 1.6.0 ，讀者根據自己版本調整。

shell中的邏輯程式碼（wordcount）:

import org.apache.spark.SparkConf

import kafka.serializer.StringDecoder

import org.apache.spark.streaming.kafka.KafkaUtils

import org.apache.spark.streaming.{Durations, StreamingContext}

val ssc = new StreamingContext(sc, Durations.seconds(5))KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, Map("bootstrap.servers" -> "master:2181,worker1:2181,worker2:2181", "metadata.broker.list" -> "master:9092,worker1:9092,worker2:9092", "group.id" -> "StreamingWordCountSelfKafkaDirectStreamScala"), Set("test")).map(t => t._2).flatMap(_.toString.split(" ")).map((_, 1)).reduceByKey(_ + _).print()ssc.start()

生產者再生產訊息：

spark streaming的反應：

返回worker2檢視消費者

可見，groupId不一樣，相互之間沒有互斥。

上述是使用 createDirectStream 方式連結kafka，實際使用中，其實就是和Receiver在api以及api中引數上有不同，其它基本一樣

參考：

http://spark.apache.org/docs/latest/streaming-kafka-integration.html