本篇文章為大家展示了怎樣解析SparkStreaming和Kafka集成的兩種方式，內(nèi)容簡明扼要并且容易理解，絕對能使你眼前一亮，通過這篇文章的詳細(xì)介紹希望你能有所收獲。

成都創(chuàng)新互聯(lián)專業(yè)為企業(yè)提供佳縣網(wǎng)站建設(shè)、佳縣做網(wǎng)站、佳縣網(wǎng)站設(shè)計(jì)、佳縣網(wǎng)站制作等企業(yè)網(wǎng)站建設(shè)、網(wǎng)頁設(shè)計(jì)與制作、佳縣企業(yè)網(wǎng)站模板建站服務(wù)，十年佳縣做網(wǎng)站經(jīng)驗(yàn)，不只是建網(wǎng)站，更提供有價值的思路和整體網(wǎng)絡(luò)服務(wù)。

Spark Streaming是基于微批處理的流式計(jì)算引擎，通常是利用Spark Core或者Spark Core與Spark Sql一起來處理數(shù)據(jù)。在企業(yè)實(shí)時處理架構(gòu)中，通常將Spark Streaming和Kafka集成作為整個大數(shù)據(jù)處理架構(gòu)的核心環(huán)節(jié)之一。

針對不同的Spark、Kafka版本，集成處理數(shù)據(jù)的方式分為兩種：Receiver based Approach和Direct  Approach，不同集成版本處理方式的支持，可參考下圖：

Receiver based Approach

基于receiver的方式是使用kafka消費(fèi)者高階API實(shí)現(xiàn)的。

對于所有的receiver，它通過kafka接收的數(shù)據(jù)會被存儲于spark的executors上，底層是寫入BlockManager中，默認(rèn)200ms生成一個block(通過配置參數(shù)spark.streaming.blockInterval決定)。然后由spark  streaming提交的job構(gòu)建BlockRdd，最終以spark core任務(wù)的形式運(yùn)行。

關(guān)于receiver方式，有以下幾點(diǎn)需要注意：

• receiver作為一個常駐線程調(diào)度到executor上運(yùn)行，占用一個cpu

• receiver個數(shù)由KafkaUtils.createStream調(diào)用次數(shù)決定，一次一個receiver

• kafka中的topic分區(qū)并不能關(guān)聯(lián)產(chǎn)生在spark streaming中的rdd分區(qū)

• 增加在KafkaUtils.createStream()中的指定的topic分區(qū)數(shù)，僅僅增加了單個receiver消費(fèi)的topic的線程數(shù)，它不會增加處理數(shù)據(jù)中的并行的spark的數(shù)量【topicMap[topic,num_threads]map的value對應(yīng)的數(shù)值是每個topic對應(yīng)的消費(fèi)線程數(shù)】

• receiver默認(rèn)200ms生成一個block，建議根據(jù)數(shù)據(jù)量大小調(diào)整block生成周期。

• receiver接收的數(shù)據(jù)會放入到BlockManager，每個executor都會有一個BlockManager實(shí)例，由于數(shù)據(jù)本地性，那些存在receiver的executor會被調(diào)度執(zhí)行更多的task，就會導(dǎo)致某些executor比較空閑

建議通過參數(shù)spark.locality.wait調(diào)整數(shù)據(jù)本地性。該參數(shù)設(shè)置的不合理，比如設(shè)置為10而任務(wù)2s就處理結(jié)束，就會導(dǎo)致越來越多的任務(wù)調(diào)度到數(shù)據(jù)存在的executor上執(zhí)行，導(dǎo)致任務(wù)執(zhí)行緩慢甚至失敗(要和數(shù)據(jù)傾斜區(qū)分開)

多個kafka輸入的DStreams可以使用不同的groups、topics創(chuàng)建，使用多個receivers接收處理數(shù)據(jù)

兩種receiver可靠的receiver：

• 可靠的receiver在接收到數(shù)據(jù)并通過復(fù)制機(jī)制存儲在spark中時準(zhǔn)確的向可靠的數(shù)據(jù)源發(fā)送ack確認(rèn)不可靠的receiver：

• 不可靠的receiver不會向數(shù)據(jù)源發(fā)送數(shù)據(jù)已接收確認(rèn)。 這適用于用于不支持ack的數(shù)據(jù)源當(dāng)然，我們也可以自定義receiver。

• receiver處理數(shù)據(jù)可靠性默認(rèn)情況下，receiver是可能丟失數(shù)據(jù)的。

• 可以通過設(shè)置spark.streaming.receiver.writeAheadLog.enable為true開啟預(yù)寫日志機(jī)制，將數(shù)據(jù)先寫入一個可靠地分布式文件系統(tǒng)如hdfs，確保數(shù)據(jù)不丟失，但會失去一定性能

限制消費(fèi)者消費(fèi)的最大速率涉及三個參數(shù)：

• spark.streaming.backpressure.enabled：默認(rèn)是false，設(shè)置為true，就開啟了背壓機(jī)制;

• spark.streaming.backpressure.initialRate：默認(rèn)沒設(shè)置初始消費(fèi)速率，第一次啟動時每個receiver接收數(shù)據(jù)的最大值;

• spark.streaming.receiver.maxRate：默認(rèn)值沒設(shè)置，每個receiver接收數(shù)據(jù)的最大速率(每秒記錄數(shù))。每個流每秒最多將消費(fèi)此數(shù)量的記錄，將此配置設(shè)置為0或負(fù)數(shù)將不會對最大速率進(jìn)行限制

在產(chǎn)生job時，會將當(dāng)前job有效范圍內(nèi)的所有block組成一個BlockRDD，一個block對應(yīng)一個分區(qū)

kafka082版本消費(fèi)者高階API中，有分組的概念，建議使消費(fèi)者組內(nèi)的線程數(shù)(消費(fèi)者個數(shù))和kafka分區(qū)數(shù)保持一致。如果多于分區(qū)數(shù)，會有部分消費(fèi)者處于空閑狀態(tài)

Direct Approach

direct approach是spark  streaming不使用receiver集成kafka的方式，一般在企業(yè)生產(chǎn)環(huán)境中使用較多。相較于receiver，有以下特點(diǎn)：

1.不使用receiver

不需要創(chuàng)建多個kafka streams并聚合它們

減少不必要的CPU占用

減少了receiver接收數(shù)據(jù)寫入BlockManager，然后運(yùn)行時再通過blockId、網(wǎng)絡(luò)傳輸、磁盤讀取等來獲取數(shù)據(jù)的整個過程，提升了效率

無需wal，進(jìn)一步減少磁盤IO操作

2.direct方式生的rdd是KafkaRDD，它的分區(qū)數(shù)與kafka分區(qū)數(shù)保持一致一樣多的rdd分區(qū)來消費(fèi)，更方便我們對并行度進(jìn)行控制

注意：在shuffle或者repartition操作后生成的rdd，這種對應(yīng)關(guān)系會失效

3.可以手動維護(hù)offset，實(shí)現(xiàn)exactly once語義

4.數(shù)據(jù)本地性問題。在KafkaRDD在compute函數(shù)中，使用SimpleConsumer根據(jù)指定的topic、分區(qū)、offset去讀取kafka數(shù)據(jù)。

但在010版本后，又存在假如kafka和spark處于同一集群存在數(shù)據(jù)本地性的問題

5.限制消費(fèi)者消費(fèi)的最大速率

spark.streaming.kafka.maxRatePerPartition：從每個kafka分區(qū)讀取數(shù)據(jù)的最大速率(每秒記錄數(shù))。這是針對每個分區(qū)進(jìn)行限速，需要事先知道kafka分區(qū)數(shù)，來評估系統(tǒng)的吞吐量。

上述內(nèi)容就是怎樣解析SparkStreaming和Kafka集成的兩種方式，你們學(xué)到知識或技能了嗎？如果還想學(xué)到更多技能或者豐富自己的知識儲備，歡迎關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道。