Kafka 資料源、Receiver 和 Direct 方式接收資料_3|學習筆記

開發者學堂課程【大資料實時計算架構 Spark 快速入門：Kafka 資料源、Receiver 和 Direct 方式接收資料_3】學習筆記，與課程緊密聯系，讓使用者快速學習知識。

課程位址：

https://developer.aliyun.com/learning/course/100/detail/1731

Kafka 資料源、Receiver 和 Direct 方式接收資料_3

不需要 Receivers：

Spark 1.3 引入了這種新的無接收端的“直接”方法，確定了更強的端到端保證。

這種方法不使用接收器來接收資料，而是定期查詢 Katka 在每個主題+分區中的最新偏移量，并相應地定義在每個批次中處理的偏移量範圍，當處理資料的作業啟動時，Kafka 的簡單消費者 AP 用于從 Kafka 讀取定義的偏移範圍(類似于從檔案系統讀取檔案)。

注意，這是 Spark 1.3 中針對 Scala 和 Java AP 引入的實驗性特性，在 Spark 1.4 中針對 Python API 引入的實驗性特性。  

與基于接收方的方法(即方法 1 )相比，這種方法有以下優點。

簡化的并行性：不需要建立多個輸入 Katka 流并将它們聯合起來。使用directStream. Spark Streaming 将建立與 Katka 分區一樣多的 RDD 分區。它們将同時從Katka讀取資料。是以 Katka 分區和 RDD 分區之間是一一對應的。這更容易了解和調整。

效率：要在第一種方法中實作零資料丢失，需要将資料存儲在 Write Ahead Log中，這将進一步複制資料。

這實際上是低效的，因為資料被有效地複制了兩次——一次由 Katka 複制，另一次由 Write Ahead Log 複制。

第二種方法消除了這個問題，因為沒有接收器，是以不需要提前寫入日志。隻要你有足夠的 Karka 保留，消息可以從 Kafka 恢複。

Exactly-once-semantics：第一種方法使用 Katka 的進階 APl 在 Zookeeper 中存儲消耗的偏移量。

這是傳統的來消費來自 Kafka 的資料。而這種方法(結合提前寫日志)可以確定零資料丢失(即至少一次語義)。在某些失敗情況下，有些記錄有可能被消耗兩次。

這是因為 Spark Streaming 可靠接收的資料與 Zookeeper 跟蹤的偏移量之間不一緻。

是以，在第二種方法中，我們使用了不使用 Zookeeper 的簡單 Kaka API。

Spark Streaming 在其檢查點内跟蹤偏移量。這消除了 Spark 流和Zookeeper/Kafka 之間的不一緻性，是以每個記錄都被 Spark 流有效地精确接收一次，盡管失敗。為了實作輸出結果的精确一次語義。将資料儲存到外部資料存儲的輸出操作必須是幂等的，或者是儲存結果和偏移量的原子事務。

注意，這種方法的一個缺點是它不會在 Zookeeper 中更新偏移量，基于 Zookeeper 的 Kafka 監控不會顯示進展。然而，你可以在每批中通路用這種方法處理的偏移量，并自己更新 Zookeeper。代碼如下：

​

directKafikaStream.transformToPair(

new Function<JavaPairRDD<String，String>，3avaPairRDD<String，String>>( ){

@Override

public JavaPairRDDeString，String ca11(3avaPairRDDeString，Strings rdd) throws Exception {

OffsetRange[] offsets - ((HasOffsetRanges) rdd.rdd().offsetRanges();

offsetRanges.set(offsets);

return rdd;

}

}

). map(

).foreachRDD(

new Function<JavaPairRDD<String，String>, void>(){

@Override

public void call(JavaPairRDD-<String，Strings rdd) throws IOException {

for (OffsetRange o : offsetRanges.get({

System. out.peintln(

o.topic( + " " + o.partition() + " " + o.fromOffset( + "  " + o.unti1offset();

);

}

return null;

}

}

);