建站
咨詢
售后
建站咨詢
新聞中心
Apache Kafka與Spark Streaming的兩種整合方法及其優(yōu)缺點
譯文
作者：陳峻編譯 2018-10-24 09:00:26
大數(shù)據(jù)
Kafka
Spark 我們在將Apache Kafka與Spark Streaming整合的實戰(zhàn)過程中，一般可以選用兩種方面來配置Spark Streaming，并接收來自Kafka的數(shù)據(jù)。第一種是利用接收器和Kafka的高級API;而第二種新的方法則并不使用接收器。這兩種方法在性能特征和語義保持上，有著不同的編程模式。
10余年的四方臺網(wǎng)站建設(shè)經(jīng)驗，針對設(shè)計、前端、開發(fā)、售后、文案、推廣等六對一服務(wù)，響應(yīng)快，48小時及時工作處理。成都全網(wǎng)營銷的優(yōu)勢是能夠根據(jù)用戶設(shè)備顯示端的尺寸不同，自動調(diào)整四方臺建站的顯示方式，使網(wǎng)站能夠適用不同顯示終端，在瀏覽器中調(diào)整網(wǎng)站的寬度，無論在任何一種瀏覽器上瀏覽網(wǎng)站，都能展現(xiàn)優(yōu)雅布局與設(shè)計，從而大程度地提升瀏覽體驗。成都創(chuàng)新互聯(lián)公司從事“四方臺網(wǎng)站設(shè)計”,“四方臺網(wǎng)站推廣”以來，每個客戶項目都認真落實執(zhí)行。
【51CTO.com快譯】Kafka與Spark Streaming的整合
我們在將Apache Kafka與Spark Streaming整合的實戰(zhàn)過程中，一般可以選用兩種方面來配置Spark Streaming，并接收來自Kafka的數(shù)據(jù)。***種是利用接收器和Kafka的高級API;而第二種新的方法則并不使用接收器。這兩種方法在性能特征和語義保持上，有著不同的編程模式。
下面讓我們來詳細探究一下這兩種方法。
一、基于接收器的方法
此法運用接收器(Receiver)來接收數(shù)據(jù)。而接收器是利用Kafka的高級消費者(consumer)API來實現(xiàn)的。此外，接收到的數(shù)據(jù)會被存儲在Spark的各個執(zhí)行器(executor)中。然后由Spark Streaming所啟動的作業(yè)來處理數(shù)據(jù)。
但是在出現(xiàn)失敗時，這種方法的默認配置可能會丟失數(shù)據(jù)。因此，我們必須在Spark Streaming中額外地啟用預(yù)寫日志(write-ahead log)，以確保數(shù)據(jù)的零丟失。它將所有接收到的Kafka數(shù)據(jù)，同步地保存到某個分布式文件系統(tǒng)的預(yù)寫日志中，以便在出現(xiàn)失敗時恢復(fù)所有的數(shù)據(jù)。
下面，我們將討論如何在Kafka-Spark Streaming應(yīng)用中，使用該基于接收器的方法。
1.鏈接
現(xiàn)在，先將您的Kafka Streaming應(yīng)用與如下的artifact相鏈接，對于Scala和Java類型的應(yīng)用，我們會用到SBT(Simple Build Tool)和Maven(一種構(gòu)建工具)的各種項目定義。
- groupId = org.apache.spark
- artifactId = spark-streaming-kafka-0-8_2.11
- version = 2.2.0
而對于Python類型的應(yīng)用，我們必須在部署自己的應(yīng)用時，額外添加上述庫、及其各種依賴項。
2.編程
隨后，我們在streaming應(yīng)用的代碼中，通過導(dǎo)入KafkaUtils，來創(chuàng)建一項DStream輸入：
- import org.apache.spark.streaming.kafka._
- val kafkaStream = KafkaUtils.createStream(streamingContext,
- [ZK quorum], [consumer group id], [per-topic number of Kafka partitions to consume])
同樣，通過使用createStream的各種變形方式，我們可以制定出不同的鍵/值類，及其對應(yīng)的解碼類。
3.部署
通常情況下，對于任何Spark應(yīng)用而言，您都可以使用spark-submit來發(fā)布自己的應(yīng)用。當(dāng)然，就具體的Scala、Java和Python應(yīng)用來說，它們在細節(jié)上會略有不同。
其中，由于Python應(yīng)用缺少SBT和Maven的項目管理，我們可以使用–packages spark-streaming-kafka-0-8_2.11、及其各個依賴項，直接添加到spark-submit處。
- ./bin/spark-submit --packages org.apache.spark:spark-streaming-kafka-0-8_2.11:2.2.0 ...
此外，我們還可以從Maven的存儲庫中下載Maven artifact的spark-streaming-Kafka-0-8-assembly所對應(yīng)的JAR包，然后使用-jars，將其添加到spark-submit處。
二、直接方法(無接收器)
在基于接收器的方法之后，新的一種無接收器式“直接”方法誕生了。此法提供了更強大的端到端保證。它定期查詢Kafka在每個topic+分區(qū)(partition)中的***偏移量，而不再使用接收器去接收數(shù)據(jù)。同時，它也定義了要在每個批次中處理的不同偏移范圍。特別是在那些處理數(shù)據(jù)的作業(yè)被啟動時，其簡單消費者(consumer)API就會被用于讀取Kafka中預(yù)定義的偏移范圍?？梢?，此過程類似于從某個文件系統(tǒng)中讀取各種文件。
注：針對Scala和Java API，Spark在其1.3版本中引入了此功能;而針對Python API，它在其1.4版本中同樣引入了該功能。
下面，我們將討論如何在Streaming應(yīng)用中使用該方法，并深入了解更多有關(guān)消費者API的鏈接：
1.鏈接
當(dāng)然，這種方法僅被Scala和Java應(yīng)用所支持，并且通過如下artifact來鏈接STB和Maven項目。
- groupId = org.apache.spark
- artifactId = spark-streaming-kafka-0-8_2.11
- version = 2.2.0
2.編程
隨后，我們在Streaming應(yīng)用的代碼中，通過導(dǎo)入KafkaUtils，來創(chuàng)建一項DStream輸入：
- import org.apache.spark.streaming.kafka._
- val directKafkaStream = KafkaUtils.createDirectStream[
- [key class], [value class], [key decoder class], [value decoder class] ](
- streamingContext, [map of Kafka parameters], [set of topics to consume])
我們必須在Kafka的參數(shù)中，指定metadata.broker.list或bootstrap.servers，以便它能夠在默認情況下，從各個Kafka分區(qū)的***偏移量開始消費。當(dāng)然，如果您在Kafka的參數(shù)中將auto.offset.reset配置為最小，那么它就會從最小的偏移開始消費。
此外，通過使用KafkaUtils.createDirectStream的各種變形方式，我們能夠從任意偏移量開始消費。當(dāng)然，我們也可以在每一個批次中，按照如下的方式去消費Kafka的偏移量。
- // Hold a reference to the current offset ranges, so downstream can use it
- var offsetRanges = Array.empty[OffsetRange]
- directKafkaStream.transform { rdd =>
- offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
- rdd
- }.map {
- ...
- }.foreachRDD { rdd =>
- for (o <- offsetRanges) {
- println(s"${o.topic} ${o.partition} ${o.fromOffset} ${o.untilOffset}")
- }
- ...
- }
如果您想使用基于Zookeeper的Kafka監(jiān)控工具(https://data-flair.training/blogs/zookeeper-in-kafka/)，來顯示Streaming應(yīng)用的進度，那么您也可以自行將其更新到Zookeeper中。
3.部署
該方面的部署過程與基于接收器的方法類似，此處就不贅述了。
三、直接方法的優(yōu)點
就Spark Streaming與Kafka整合的角度而言，第二種方法較***種方法有著如下的優(yōu)點：
1.簡化并行
無需創(chuàng)建與合并多個輸入的Kafka Streams(https://data-flair.training/blogs/kafka-streams/)。但是，Sparking Streaming會創(chuàng)建同樣多的RDD(Resilient Distributed Datasets，彈性分布式數(shù)據(jù)集)分區(qū)，以供多個Kafka分區(qū)使用直接的方法進行消費。這些分區(qū)也會并行地從Kafka中讀取數(shù)據(jù)。因此我們可以說：在Kafka和RDD分區(qū)之間存在更容易被理解和調(diào)整的、一對一的映射關(guān)系。
2.效率
為了實現(xiàn)數(shù)據(jù)的零丟失，***種方法需要將數(shù)據(jù)存儲在預(yù)寫日志中，以供進一步復(fù)制數(shù)據(jù)。此方法的效率實際上是比較低的，因為數(shù)據(jù)被Kafka和預(yù)寫日志實際復(fù)制了兩次。而在直接的方法中，由于沒有了接收器，因此不需要預(yù)先寫入日志，此問題也就迎刃而解了。只要您擁有足夠多的Kafka數(shù)據(jù)保留，各種消息就能夠從Kafka中被恢復(fù)回來。
3.準(zhǔn)確到位的語義
在***種方法中，我們使用Kafka的高級API，在Zookeeper中存儲被消費的偏移量。然而，這種傳統(tǒng)的、從Kafka中消費數(shù)據(jù)的方式，雖然能夠確保數(shù)據(jù)的零丟失，但是在某些失敗情況下，數(shù)據(jù)可能會被小概率地消費兩次。實際上，這種情況源自那些被Spark Streaming可靠地接收到的數(shù)據(jù)，與Zookeeper跟蹤到的偏移量之間所產(chǎn)生的不一致。因此在第二種方法中，我們不再使用Zookeeper，而只是使用一個簡單的Kafka API。Spark Streaming通過其各個檢查點(checkpoints)，來跟蹤不同的偏移量，籍此消除了Spark Streaming和Zookeeper之間的不一致性。
可見，就算出現(xiàn)了失敗的情況，那些記錄也都會被Spark Streaming有效地、準(zhǔn)確地一次性接收。它能夠確保我們的輸出操作，即：將數(shù)據(jù)保存到外部數(shù)據(jù)存儲庫時，各種保存結(jié)果和偏移量的冪等性、和原子事務(wù)性，這同時也有助于實現(xiàn)準(zhǔn)確到位的語義。
不過，這種方法也有一個缺點：由于它不會在Zookeeper中更新各種偏移量，因此那些基于Zookeeper的Kafka監(jiān)控工具將無法顯示進度。當(dāng)然，您也可以自行訪問每個批次中由此方法處理的偏移量，并更新到Zookeeper之中。
結(jié)論
通過上述討論，我們學(xué)到了Kafka與Spark Streaming整合的全體概念。同時，我們也討論了Kafka-Spark Streaming的兩種不同配置方法：接收器方法和直接方法，以及直接方法的幾項優(yōu)點。
原文標(biāo)題：Apache Kafka + Spark Streaming Integration，作者：Rinu Gour
【51CTO譯稿，合作站點轉(zhuǎn)載請注明原文譯者和出處為51CTO.com】
文章名稱：ApacheKafka與SparkStreaming的兩種整合方法及其優(yōu)缺點
新聞來源：http://www.5511xx.com/article/dpisdpp.html
