Spark平臺(tái)中Kafka偏移量的讀取管理與設(shè)計(jì)

高宗寶　劉麗美　張家銘　宋國(guó)興

摘? 要： 隨著移動(dòng)互聯(lián)網(wǎng)技術(shù)的大規(guī)模發(fā)展，創(chuàng)新型互聯(lián)網(wǎng)公司和迭代型各行各業(yè)應(yīng)用產(chǎn)品層出不窮，門(mén)戶訪問(wèn)、好友互動(dòng)等操作產(chǎn)生的大規(guī)模日志記錄，對(duì)大數(shù)據(jù)處理的實(shí)時(shí)性、準(zhǔn)確性和高可用性發(fā)起了挑戰(zhàn)。Kafka是一種高吞吐量分布式發(fā)布訂閱消息系統(tǒng)，其在高并發(fā)數(shù)據(jù)讀寫(xiě)方面優(yōu)勢(shì)明顯，但其提供的數(shù)據(jù)消費(fèi)方式存在數(shù)據(jù)丟失和重復(fù)的風(fēng)險(xiǎn)。本文首先介紹Kafka架構(gòu)及其Offset管理，介紹了新型流式數(shù)據(jù)處理框架SparkStreaming與Kafka的結(jié)合，并說(shuō)明了Kafka數(shù)據(jù)消費(fèi)方面存在的缺陷，最后提出了一種基于SparkStreaming讀取Kafka的近似Exactly Once方案實(shí)現(xiàn)。通過(guò)搭建實(shí)驗(yàn)環(huán)境進(jìn)行對(duì)比測(cè)試，驗(yàn)證了該設(shè)計(jì)可以在保證數(shù)據(jù)讀取效率的前提下確保數(shù)據(jù)的準(zhǔn)確性。

關(guān)鍵詞： Kafka;Offset;SparkStreaming;數(shù)據(jù)準(zhǔn)確性

中圖分類號(hào)： TP302? ? 文獻(xiàn)標(biāo)識(shí)碼： A? ? DOI：10.3969/j.issn.1003-6970.2019.07.022

【Abstract】： With the large-scale development of mobile Internet technology， the application products of various industries emerge in an endless stream. The large-scale log records generated by portal access， friend interaction and other operations challenge the real-time， accuracy and high availability of large data processing. Kafka is a high throughput distributed publish-subscribe messaging system， which has obvious advantages in high concurrent data reading and writing， but its data consumption mode has the risk of data loss and duplication. Firstly， this paper introduces Kafka architecture and its Offset management， introduces the combination of SparkStreaming and Kafka， a new streaming data processing framework， and illustrates the shortcomings of Kafka data consumption. Finally， an approximate Exactly One scheme based on SparkStreaming to read Kafka is proposed. By building an experimental environment for comparative testing， it is verified that the design can ensure the accuracy of data on the premise of ensuring the efficiency of data reading.

【Key words】： Kafka; Offset; SparkStreaming; Data accuracy

0? 引言

隨著IT和移動(dòng)互聯(lián)網(wǎng)技術(shù)的飛速發(fā)展，互聯(lián)網(wǎng)[1]軟件產(chǎn)品迭代開(kāi)發(fā)、層出不窮，數(shù)據(jù)量激增，如何存儲(chǔ)和及時(shí)處理這些海量數(shù)據(jù)，挖掘其中企業(yè)比較感興趣的價(jià)值信息（如用戶喜好等）進(jìn)而提供更好的產(chǎn)品服務(wù)（如好友推薦、產(chǎn)品推廣等）是數(shù)據(jù)導(dǎo)向型公司迫切需要解決的問(wèn)題。門(mén)戶網(wǎng)站訪問(wèn)、好友聊天、支付交易記錄等用戶操作產(chǎn)生的大規(guī)模日志記錄，對(duì)大數(shù)據(jù)處理的實(shí)時(shí)性和高并發(fā)性發(fā)起了挑戰(zhàn)。傳統(tǒng)的數(shù)據(jù)存儲(chǔ)介質(zhì)，如關(guān)系型數(shù)據(jù)庫(kù)、文件系統(tǒng)等無(wú)法滿足數(shù)據(jù)實(shí)時(shí)讀寫(xiě)傳輸和流處理，Apache Kafka應(yīng)運(yùn)而生。Kafka是由Apache軟件基金會(huì)開(kāi)發(fā)的一個(gè)開(kāi)源流處理平臺(tái)[2]，是主要用Scala編寫(xiě)的一種高吞吐量分布式發(fā)布訂閱消息系統(tǒng)，因其擴(kuò)展性好、高吞吐量、快速持久化、高可用性等優(yōu)點(diǎn)被各大消息系統(tǒng)、日志分析平臺(tái)、流數(shù)據(jù)處理平臺(tái)、門(mén)戶網(wǎng)站等廣泛使用。

1? Kafka簡(jiǎn)介

1.1? Kafka架構(gòu)

Kafka消息系統(tǒng)的基本架構(gòu)如圖1所示。其架構(gòu)主要包括以下幾個(gè)組件：

（1）Message：消息，通信基本單位。

（2）Broker：Kafka節(jié)點(diǎn)實(shí)例，對(duì)應(yīng)為Kafka集群的一臺(tái)機(jī)器。

（3）Topic：主題，表示Kafka數(shù)據(jù)處理的消息源，數(shù)據(jù)的讀寫(xiě)都要指定主題。

（4）Producer：數(shù)據(jù)生產(chǎn)者，向某個(gè)Topic發(fā)布消息的對(duì)象，即一種push操作，將消息推送給代理對(duì)象Broker進(jìn)行存儲(chǔ)。

（5）Consumer：數(shù)據(jù)消費(fèi)者，訂閱某個(gè)Topic并處理消息的對(duì)象，即一種pull操作，主動(dòng)拉去數(shù)據(jù)，Consumer自己控制消息的讀取速度和數(shù)量，如果Topic中沒(méi)有數(shù)據(jù)，那么會(huì)周期性的pull操作直到有數(shù)據(jù)產(chǎn)生。

（6）Partition：分區(qū)，一個(gè)Topic可以有多個(gè)partition，一個(gè)消息實(shí)際存儲(chǔ)在Topic的某個(gè)Partition中，每個(gè)Parition可以保證消息的有序性。

（7）Replications：分區(qū)副本，每個(gè)分區(qū)都可以設(shè)定副本數(shù)目分布到不同的Broker中以便于容錯(cuò)。

（8）ConsumerGroup：消費(fèi)者組，一組consumer的集合，group訂閱的某個(gè)topic下的每個(gè)分區(qū)只能被其中的一個(gè)consumer消費(fèi)，不會(huì)出現(xiàn)一個(gè)分區(qū)的數(shù)據(jù)被同一個(gè)group下的多個(gè)consumer消費(fèi)的情況，可以理解為ConsumerGroup是Kafka提供的可擴(kuò)展且具有容錯(cuò)性的消費(fèi)者機(jī)制，在開(kāi)發(fā)過(guò)程中使用group.id來(lái)標(biāo)識(shí)。

Kafka集群中的所有節(jié)點(diǎn)都是平等的，不采用Master-Slave結(jié)構(gòu)，這樣就不會(huì)出現(xiàn)類似HDFS的單點(diǎn)故障問(wèn)題。Kafka利用zookeeper來(lái)解決分布式一致性問(wèn)題，將broker節(jié)點(diǎn)、topic元數(shù)據(jù)信息等全部存儲(chǔ)到zookeeper中。

為了保證較高的讀寫(xiě)效率，對(duì)于每個(gè)partition，消息讀寫(xiě)都有一個(gè)固定的副本完成，即Leader節(jié)點(diǎn)，其他的副本是Follower節(jié)點(diǎn)。Follower節(jié)點(diǎn)會(huì)定期同步Leader節(jié)點(diǎn)的數(shù)據(jù)。

當(dāng)使用工具kafka-topics.sh創(chuàng)建topic后，kafka會(huì)根據(jù)選舉策略對(duì)每個(gè)partition都選出一個(gè)Leader節(jié)點(diǎn)和相應(yīng)數(shù)量的Follower節(jié)點(diǎn)（通過(guò)參數(shù)replication- factor控制副本數(shù)量）。圖2描述的是創(chuàng)建主題t1，分區(qū)數(shù)量為4，副本數(shù)量為3的情況。

以partition=1為例，其讀寫(xiě)節(jié)點(diǎn)是437（broker.id），副本節(jié)點(diǎn)分別是437、441、436，副本同步隊(duì)列分別是436、440、441。ISR（in-sync replica，副本同步隊(duì)列）是由Leader維護(hù)的與主節(jié)點(diǎn)數(shù)據(jù)同步的一個(gè)節(jié)點(diǎn)集合，當(dāng)producer發(fā)送消息到leader后，follower會(huì)同步消息，如果某個(gè)follower沒(méi)有同步

leader的消息太多或者失效，那么leader會(huì)將其從ISR中剔除。

當(dāng)leader失效后，kafka會(huì)從ISR中的副本中選舉出新的leader以保證服務(wù)的可用。

1.2? 讀寫(xiě)offset管理

Topic可以簡(jiǎn)單理解為一個(gè)queue，消息的生產(chǎn)與消費(fèi)都要聲明消息所在的queue。為了提高數(shù)據(jù)讀寫(xiě)效率和數(shù)據(jù)吞吐量，在物理上Topic被分成了多個(gè)partition，每個(gè)partition表示一個(gè)文件夾，命名為“topic名-分區(qū)號(hào)”，每個(gè)文件夾中保存消息數(shù)據(jù)、消息索引等。

任何發(fā)布到partition的消息都會(huì)被append到文件尾部，每條消息在文件中的存儲(chǔ)位置稱之為偏移量（offset，long型整數(shù)），通過(guò)partition+offset可以唯一標(biāo)識(shí)一條消息。因?yàn)槭亲芳硬僮?，所以在partition中消息是有序?qū)懭氪疟P(pán)的，其寫(xiě)入和索引讀取效率都很高。圖3表明了一個(gè)分區(qū)數(shù)量為3的topic消息寫(xiě)入狀態(tài)。

當(dāng)消息寫(xiě)入時(shí)，kafka會(huì)按照默認(rèn)規(guī)則規(guī)定消息會(huì)被寫(xiě)入到哪個(gè)partition中。如果自定義規(guī)則合理，那么可以保證消息被均勻地分布到broker中。

可以看出，消息的消費(fèi)，核心是對(duì)partition和offset的管理。Kafka由ConsumerGroup控制消息的消費(fèi)和偏移量，而不是交給Broker去存儲(chǔ)，甚至可以加以控制回到一個(gè)之前的偏移位置再次消費(fèi)消息。

Kafka提供了自動(dòng)和手動(dòng)2種偏移量管理方式[4，5]。

Kafka默認(rèn)會(huì)定期自動(dòng)提交偏移信息，即enable. auto.commit=true。在kafka0.10版本之前，offset信息提交到zk中保存，但由于zk不適合大批量數(shù)據(jù)的并行讀寫(xiě)操作，自kafka0.10版本，offset信息自動(dòng)提交到名為_(kāi)_consumer_offsets的topic存儲(chǔ)。該topic默認(rèn)有50個(gè)分區(qū)，保存了每個(gè)ConsumerGroup消費(fèi)的Topic所有partition的offset信息，如圖4所示。

當(dāng)然也可以采用手動(dòng)更新的方法提交offset。

在消息消費(fèi)過(guò)程中，Kafka提供了如下3種可能的傳輸保障（consumer delivery guarantee）。

（1）At most once：這種模式下，消息可能會(huì)丟，但是絕對(duì)不會(huì)重復(fù)消費(fèi)。如果consumer設(shè)定autocommit偏移量，consumer在讀取到數(shù)據(jù)后立即更新offset后未來(lái)得及處理消息（如consumer系統(tǒng)崩潰），下次重新工作時(shí)無(wú)法讀取之前未處理的消息，導(dǎo)致數(shù)據(jù)丟失。

（2）At least once：這種模式下，數(shù)據(jù)不會(huì)丟失，但是可能會(huì)存在重復(fù)消費(fèi)。consumer在讀取到數(shù)據(jù)后立即處理，處理完成后沒(méi)來(lái)得及提交偏移量。下次重新工作時(shí)還會(huì)重新讀取已處理但是沒(méi)有提交偏移量的數(shù)據(jù)，導(dǎo)致數(shù)據(jù)重復(fù)。

（3）Exactly once：這種模式下，數(shù)據(jù)既不會(huì)丟失也不會(huì)重復(fù)消費(fèi)，需要協(xié)調(diào)消費(fèi)數(shù)據(jù)和offset進(jìn)行精確事務(wù)管理，如將數(shù)據(jù)和offset信息寫(xiě)入到HDFS等外部介質(zhì)中，這種模式對(duì)處理效率有一定影響。

2? SparkStreaming簡(jiǎn)介

SparkStreaming是基于spark的流式批處理引擎，可以實(shí)現(xiàn)高吞吐量的、具備容錯(cuò)機(jī)制的實(shí)時(shí)流數(shù)據(jù)的處理，能夠與RDD算子、機(jī)器學(xué)習(xí)、SparkSQL以及圖形圖像處理框架無(wú)縫連接[3-6]。類似于Apache Storm，用于流式數(shù)據(jù)的處理。SparkStreaming支持多種數(shù)據(jù)源，如Flume、Kafka、HDFS、套接字等，經(jīng)過(guò)一系列RDD算子或windows等高級(jí)函數(shù)進(jìn)? 行處理后，將結(jié)果寫(xiě)入到文件系統(tǒng)、數(shù)據(jù)庫(kù)等輸出源中。

SparkStreaming接收實(shí)時(shí)數(shù)據(jù)流，并以某一時(shí)間間隔（batchDuration）劃分為一個(gè)個(gè)數(shù)據(jù)批次（batch）交給Spark Engine處理。SparkStreaming的數(shù)據(jù)處理流程如圖5所示。

Dstream是SparkStreaming中特有的數(shù)據(jù)類型，表示一系列連續(xù)的RDD集合，即數(shù)據(jù)批次集合，存儲(chǔ)方式是Map，對(duì)每個(gè)批次數(shù)據(jù)的處理實(shí)際上是RDD的操作，每個(gè)批次的處理邏輯是完全相同的。

SparkStreaming+Kafka進(jìn)行流數(shù)據(jù)處理被廣泛采用，本文后續(xù)討論基于spark2.3+kafka0.10展開(kāi)。

3? 一種可靠的Kafka消費(fèi)方案

3.1? 方案設(shè)計(jì)

SparkStreaming通過(guò)KafkaUtils.createDirectStream創(chuàng)建數(shù)據(jù)流Dstream，默認(rèn)情況下enable.auto.commit= true自動(dòng)提交offset，即對(duì)應(yīng)At most once模式。并且無(wú)論StreamingContext是否安全終止，都會(huì)出現(xiàn)在一段時(shí)間后已消費(fèi)offset值等于最新offset值，盡管此時(shí)數(shù)據(jù)還遠(yuǎn)沒(méi)有消費(fèi)完數(shù)據(jù)。具體見(jiàn)方案測(cè)試。

設(shè)置enable.auto.commit=false可以手動(dòng)提交offset更新，如Spark中可通過(guò)stream.asInstance-Of[CanCommitOffsets].commitAsync （offsetRanges）來(lái)進(jìn)行數(shù)據(jù)處理完后手動(dòng)提交更新。需要注意的是，此方法將offsetRanges保存在一個(gè)隊(duì)列中，只有等consumer獲取下一批次數(shù)據(jù)后才提交offsetRanges。方案測(cè)試中通過(guò)5次實(shí)驗(yàn)對(duì)比進(jìn)行驗(yàn)證。具體見(jiàn)方案測(cè)試。

在很多設(shè)計(jì)方案中將offset更新到zk中存儲(chǔ)。然而zk并不適合大規(guī)模數(shù)據(jù)并發(fā)讀寫(xiě)，尤其是寫(xiě)效率不高。Kafka允許多個(gè)ConsumerGroup并行讀寫(xiě)數(shù)據(jù)，如果offset全部在zk中管理會(huì)影響zk性能，進(jìn)而影響kafka的leader選舉、集群同步等功能。

因此，綜合考慮kafka集群性能和數(shù)據(jù)讀寫(xiě)效率，本文設(shè)計(jì)實(shí)現(xiàn)了一種At least Once方案SEO （Similar Exactly Once），每個(gè)ConsumerGroup在本地系統(tǒng)中維護(hù)offset信息，KafkaCluster提供維護(hù)信息，在不影響讀取效率的情況下趨向于Exactly Once保障。

SEO方案實(shí)現(xiàn)的假設(shè)條件是zk不可靠或存在延遲，實(shí)現(xiàn)目的是數(shù)據(jù)不可丟失，極端情況下允許數(shù)據(jù)重復(fù)。方案的一些專有名詞包括：

客戶端：運(yùn)行SparkStreaming程序所在的機(jī)器;

gtoffset文件：客戶端存儲(chǔ)的偏移量文件，文件存儲(chǔ)路徑類似于...groupid/topicname/gtoffset，文件包括groupid消費(fèi)topicname所有分區(qū)的offset信息。

偏移量越界：包括低越界、高越界。低越界指的是gtoffset記錄的偏移量信息小于Kafka目前可用的offset最小值，高越界指的是gtoffset記錄的偏移量信息超過(guò)Kafka目前最新的offset值。

方案的實(shí)現(xiàn)思路如下。

（1）在客戶端是否存在gtoffset文件，若不存在，說(shuō)明groupid是第一次消費(fèi)Topic，那么按照auto.offset.reset=earliest從當(dāng)前可用的最小offset讀取數(shù)據(jù);如果存在，說(shuō)明groupid已經(jīng)消費(fèi)過(guò)Topic，讀取得到offset集合A。

（2）使用spark-streaming-kafka-0-8中的Kafka?Cluster構(gòu)建Kafka集群連接，進(jìn)行偏移量越界判斷。使用getEarliestLeaderOffsets得到Topic的最小可用offset集合M，使用getLatestLeaderOffsets得到Topic的最大可用offset集合N。

（3）如果A中所有分區(qū)的offset都滿足offset_ （M，par）≤offset_（A，par）≤offset_（N，par）那么說(shuō)明A有效，A不需要更新;如果A中存在分區(qū)的offset滿足offset_（M，par）≥offset_（A，par），即A中有的分區(qū)offset比最小值都小，低越界，那么更新這些offset為M中對(duì)應(yīng)分區(qū)的offset;同樣道理，如果A中存在分區(qū)的offset滿足offset_（A，par）≥offset_（N，par），即A中有的分區(qū)offset比最大值都大，高越界，那么更新這些offset為N重對(duì)應(yīng)分區(qū)的offset。

（4）解決偏移量越界后，使用更新后A集合拉取Kafka中的數(shù)據(jù)進(jìn)行處理，處理成功后將最新offset信息寫(xiě)入到gtoffset文件中。因?yàn)閛ffset更新到本地文件，無(wú)需與zk、kafka等建立外部連接，可以保證更新效率，程序異常也可控制，所以該方案可以類似實(shí)現(xiàn)Exactly once傳輸保障。

3.2? 方案測(cè)試

本次測(cè)試共包括3次試驗(yàn)，3次實(shí)驗(yàn)環(huán)境完成相同，軟硬件環(huán)境如下。

第一次實(shí)驗(yàn)為enable.auto.commit=true，此時(shí)存在數(shù)據(jù)丟失情況，且出現(xiàn)offset更新為最大值的bug。實(shí)驗(yàn)過(guò)程是：topic=test1共4個(gè)分區(qū)，寫(xiě)入100006條記錄，設(shè)定程序時(shí)間間隔為2 s，每秒每分區(qū)最大讀取50條記錄，過(guò)10 s時(shí)間后停止spark程序，此時(shí)數(shù)據(jù)沒(méi)有處理完，但是已消費(fèi)offset（CURRENT-OFFSET）已達(dá)到最大值（LOG-END- OFFSET），具體結(jié)果見(jiàn)圖6。再次啟動(dòng)程序后沒(méi)有數(shù)據(jù)可讀，數(shù)據(jù)丟失。經(jīng)過(guò)10次修改時(shí)間間隔和處理數(shù)據(jù)條數(shù)，都復(fù)現(xiàn)同樣的問(wèn)題。

第二次實(shí)驗(yàn)為通過(guò)CanCommitOffsets手動(dòng)提交偏移量，共包括5次驗(yàn)證。實(shí)驗(yàn)過(guò)程是：topic=test2共4個(gè)分區(qū)，寫(xiě)入100000條記錄，測(cè)試5次，每次修改批次間隔和每分區(qū)每秒最大讀取消息數(shù)，每次在消費(fèi)過(guò)程中終止spark程序一次，然后重啟程序直到消費(fèi)完數(shù)據(jù)。得到的實(shí)驗(yàn)結(jié)果如表1。

通過(guò)實(shí)驗(yàn)結(jié)果發(fā)現(xiàn)每次均存在重復(fù)消費(fèi)，重復(fù)消費(fèi)的數(shù)量等于分區(qū)數(shù)、間隔時(shí)間、每分區(qū)每秒最大消息數(shù)三者的乘積（假設(shè)在消費(fèi)過(guò)程中只有一次終止）。

第三次實(shí)驗(yàn)為通過(guò)SEO方案手動(dòng)提交偏移量，共包括5次驗(yàn)證。實(shí)驗(yàn)過(guò)程是：topic=test3共4個(gè)分區(qū)，寫(xiě)入100000條記錄，測(cè)試5次，每次修改批次間隔和每分區(qū)每秒最大讀取消息數(shù)，每次在消費(fèi)過(guò)程中通過(guò)ssc.stop（true， true）安全終止spark流程序一次，然后重啟程序直到消費(fèi)完數(shù)據(jù)。得到的實(shí)驗(yàn)結(jié)果如表2。

通過(guò)實(shí)驗(yàn)結(jié)果發(fā)現(xiàn)每次均不存在重復(fù)消費(fèi)也不存在數(shù)據(jù)丟失，整個(gè)實(shí)現(xiàn)過(guò)程中沒(méi)有頻繁與第三方數(shù)據(jù)源進(jìn)行交互，達(dá)到了數(shù)據(jù)不丟失的目的，近似實(shí)現(xiàn)了Exactly Once模式。當(dāng)然，在極端情況下，如果某個(gè)批次數(shù)據(jù)已經(jīng)處理結(jié)束（如導(dǎo)入到數(shù)據(jù)庫(kù)中）后，即使安全終止spark任務(wù)也未能更新本地gtoffset文件，此時(shí)重啟spark任務(wù)會(huì)出現(xiàn)數(shù)據(jù)重復(fù)消費(fèi)的問(wèn)題。

4? 結(jié)束語(yǔ)

互聯(lián)網(wǎng)飛速發(fā)展，數(shù)據(jù)質(zhì)量和數(shù)據(jù)價(jià)值最大化是每個(gè)互聯(lián)網(wǎng)企業(yè)和傳統(tǒng)企業(yè)都需要考慮的問(wèn)題，數(shù)據(jù)存儲(chǔ)與計(jì)算的并發(fā)性、實(shí)時(shí)性導(dǎo)致的產(chǎn)品性能優(yōu)劣直接影響了用戶的體驗(yàn)。本文首先介紹新型流式數(shù)據(jù)處理框架SparkStreaming與Kafka的數(shù)據(jù)消費(fèi)結(jié)合，提出了一種基于SparkStreaming讀取Kafka的近似Exactly Once方案實(shí)現(xiàn)并搭建集群環(huán)境繼續(xù)數(shù)據(jù)準(zhǔn)確性驗(yàn)證。

參考文獻(xiàn)

[1] 趙旭劍， 鄧思遠(yuǎn)， 李波， 等. 互聯(lián)網(wǎng)新聞話題特征選擇與構(gòu)建[J]. 軟件， 2015， 36（7）： 17-20.

[2] Wang J， Wang W， Chen R. Distributed Data Streams Processing Based on Flume/Kafka/Spark[C]//International Conference on Mechatronics and Industrial Informatics. 2015.

[3] Ichinose A， Takefusa A， Nakada H， et al. A study of a video analysis framework using Kafka and spark streaming[C]// IEEE International Conference on Big Data. IEEE， 2017： 2396-2401.

[4] 王巖， 王純. 一種基于Kafka的可靠的Consumer的設(shè)計(jì)方案[J]. 軟件， 2016， 37（1）： 61-66.

[5] 王鄭合， 王鋒， 鄧輝， 等. 一種優(yōu)化的Kafka消費(fèi)者/客戶端負(fù)載均衡算法[J]. 計(jì)算機(jī)應(yīng)用研究， 2017， 34（8）： 2306-2309.

[6] 鄭健， 馮瑞. 基于Spark的實(shí)時(shí)視頻分析系統(tǒng)[J]. 計(jì)算機(jī)系統(tǒng)應(yīng)用， 2017， （12）. doi：10.15888/j.cnki.csa.006112.