基于改進(jìn)的Flume的實(shí)時(shí)數(shù)據(jù)采集系統(tǒng)

朱濤 孫知信 宮婧

摘? 要：任意一個(gè)分布式系統(tǒng)都必須滿足CAP理論，在數(shù)據(jù)分析分析系統(tǒng)中，最為重要的是效率以及可靠性，而數(shù)據(jù)采集時(shí)整個(gè)分析系統(tǒng)的基石，構(gòu)建基于改進(jìn)的Flume的實(shí)時(shí)數(shù)據(jù)采集系統(tǒng)，通過flume采集數(shù)據(jù)，采用復(fù)合型Channel與flume相結(jié)合，在保證數(shù)據(jù)源的豐富性和可靠性的前提下，提高采集的效率。實(shí)驗(yàn)結(jié)果表明，該系統(tǒng)的各項(xiàng)功能符合預(yù)期結(jié)果，F(xiàn)lume使用復(fù)合型Channel可以提高采集效率。

關(guān)鍵詞：Flume? CAP? 實(shí)時(shí)性? 分布式

中圖分類號(hào)：TP273.5? ? ? ? ? ? ? ? ? ? ? ?文獻(xiàn)標(biāo)識(shí)碼：A文章編號(hào)：1672-3791（2021）04（b）-0073-04

Real Time Data Acquisition System Based on Improved Flume

ZHU Tao? ?SUN Zhixin*? GONG Jing

（Nanjing University of Posts and Telecommunications， Nanjing， Jiangsu Province， 210023 China）

Abstract：? Any distributed system must meet the CAP theory. In a data analysis system， the most important thing is efficiency and reliability. The cornerstone of the entire analysis system for data collection is to build a real-time data collection system based on improved Flume. Collect data through flume， and use the combination of composite channel and flume to improve the efficiency of collection while ensuring the richness and reliability of the data source. The experimental results show that the functions of the system meet the expected results. Flume uses the composite channel can improve the collection efficiency.

Key Words： Flume; CAP; Real-time; Distributed

隨著互聯(lián)網(wǎng)技術(shù)飛速發(fā)展，各種網(wǎng)絡(luò)應(yīng)用呈現(xiàn)爆發(fā)式增長，同時(shí)用戶使用量也急劇增加，這也意味著各種智能終端產(chǎn)生的日志數(shù)據(jù)與日俱增，如何更好地管理以及更快地采集日志數(shù)據(jù)，成了一個(gè)亟待解決的問題，而數(shù)據(jù)是實(shí)現(xiàn)大數(shù)據(jù)研究的基礎(chǔ)，傳統(tǒng)的數(shù)據(jù)采集技術(shù)方案己經(jīng)難以滿足快速采集高質(zhì)量的數(shù)據(jù)集的需求。

1? 相關(guān)技術(shù)研

hossam hakeem[1]在基于對(duì)當(dāng)前的國內(nèi)外的大數(shù)據(jù)環(huán)境進(jìn)行的深度剖析的基礎(chǔ)之上，提出基于大數(shù)據(jù)的數(shù)據(jù)分析的軟件架構(gòu)，并針對(duì)于自己需要處理的數(shù)據(jù)類型提出一種基于分層的數(shù)據(jù)分析的模型。尚凱[2]基于對(duì)國內(nèi)運(yùn)營商數(shù)據(jù)的復(fù)雜性以及數(shù)據(jù)采集時(shí)的困難的研究，提出一種新的方案，其規(guī)定數(shù)據(jù)采集的來源均來源于企業(yè)B域、O域、M域以及企業(yè)外部，采用高可用、高可靠、分布式的數(shù)據(jù)采集方法，并對(duì)這些數(shù)據(jù)進(jìn)行清洗、處理后，將數(shù)據(jù)存儲(chǔ)于運(yùn)營商的數(shù)據(jù)庫當(dāng)中，此方案在對(duì)于海量的結(jié)構(gòu)化以及非結(jié)構(gòu)化的數(shù)據(jù)進(jìn)行采集時(shí)可以提供很好的實(shí)時(shí)性。

于秦[3]通過開源軟件Flume設(shè)計(jì)實(shí)現(xiàn)一款分布式多平臺(tái)多系統(tǒng)收集多種日志的系統(tǒng)，此系統(tǒng)具有高吞吐量、可擴(kuò)展性強(qiáng)、高聚合等特征。在提供高速數(shù)據(jù)采集解決方案方面， 李祥等[4]研究者中同時(shí)使用這兩個(gè)組件，系統(tǒng)的整體架構(gòu)分為數(shù)據(jù)采集層、數(shù)據(jù)分析層、Web界面展示組成，而數(shù)據(jù)分析采用了Hadoop和Strom分別實(shí)現(xiàn)數(shù)據(jù)的離線和實(shí)時(shí)計(jì)算，數(shù)據(jù)采集層采用Flume來實(shí)時(shí)地采集數(shù)據(jù)。陳飛等[5]研究者采用Flume的數(shù)據(jù)采集框架和ElasticSearch組合來對(duì)Nginx的日志進(jìn)行數(shù)據(jù)采集，并對(duì)這些數(shù)據(jù)進(jìn)行分析，從而完成對(duì)整個(gè)系統(tǒng)的實(shí)現(xiàn)。Hadoop是一般是用作離線處理的，而Strom則是一個(gè)很好的實(shí)時(shí)計(jì)算框架。

通過上述分析，該文將構(gòu)建基于改進(jìn)的Flume的實(shí)時(shí)數(shù)據(jù)采集系統(tǒng)，在提高數(shù)據(jù)的采集效率同時(shí)采用分布式系統(tǒng)滿足高可用性，提高了分析系統(tǒng)的分析效率。

2? 基于改進(jìn)的Flume的實(shí)時(shí)數(shù)據(jù)采集系統(tǒng)

現(xiàn)如今的研究更多地關(guān)注動(dòng)態(tài)實(shí)時(shí)數(shù)據(jù)，而不是靜態(tài)數(shù)據(jù)，這提出了更高的技術(shù)要求。要處理流數(shù)據(jù)，第一步是收集大規(guī)模實(shí)時(shí)流數(shù)據(jù)[6]。由于實(shí)時(shí)數(shù)據(jù)傳輸?shù)牟环€(wěn)定性，流數(shù)據(jù)的收集與傳統(tǒng)方法大不相同。隨著數(shù)據(jù)種類變得復(fù)雜且框架自帶的Memory Channel和File Channel都會(huì)有各種各樣的問題，無法提高更好地實(shí)時(shí)數(shù)據(jù)采集[7]。下面提出對(duì)Flume框架的改進(jìn)。

Flume自身提供了Memory Channel和File Channel。當(dāng)相關(guān)人員使用Memory Channel，它會(huì)將events存放于內(nèi)存的隊(duì)列中保證它的效率，它的優(yōu)點(diǎn)就是高效、高吞吐量，但是缺點(diǎn)也很明顯，當(dāng)機(jī)器宕機(jī)或者服務(wù)死掉的時(shí)候，內(nèi)存中的數(shù)據(jù)都會(huì)丟失，從而導(dǎo)致分析得不到準(zhǔn)確的數(shù)據(jù)。而File Channel卻相反，它是將所有的events被保存在本地的磁盤文件中，優(yōu)點(diǎn)是容量較大且發(fā)生故障時(shí)數(shù)據(jù)可恢復(fù)，缺點(diǎn)就是速度較慢。

為了充分利用這兩種channel的優(yōu)勢(shì)，該文引入復(fù)合型channel，復(fù)合型channel根據(jù)其內(nèi)在的每個(gè)channel的使用情況以及下游Sink的處理情況，來自主地選擇下次使用哪種channel來進(jìn)行數(shù)據(jù)的存儲(chǔ)。Sink的接收速度很快，當(dāng)Sink接收器處理速度夠快，并且Channel沒有存放過多數(shù)據(jù)的時(shí)候，相關(guān)人員可以采用Memory Channel，從而使得整個(gè)的系統(tǒng)的傳輸效率最大化;反之，當(dāng)Sink接收器處理速度跟不上，同時(shí)希望Channel可以暫時(shí)存儲(chǔ)采集到的數(shù)據(jù)時(shí)，相關(guān)人員可以采用File Channel來減少下游數(shù)據(jù)處理的壓力。因此，需要實(shí)現(xiàn)complex Channel，即就能智能地在兩個(gè)Channel之間切換。復(fù)合channel的運(yùn)行流程圖見圖1。

步驟1：flume從磁盤實(shí)時(shí)采集數(shù)據(jù)，經(jīng)過攔截器對(duì)數(shù)據(jù)進(jìn)行一些預(yù)處理。

步驟2：經(jīng)過預(yù)處理的數(shù)據(jù)通過選擇器，選擇復(fù)合型的channel。

步驟3：定義兩個(gè)原子布爾型變量（putToMemCh-

annel、takeFromMemChannel）作為標(biāo)志位，分別表示是否寫入MemChannel和是否從MemChannel中取出。

步驟4：執(zhí)行put方法，判斷是否可以往內(nèi)存MemChannel寫入數(shù)據(jù)，如果可以，則轉(zhuǎn)到步驟5，否則，轉(zhuǎn)到步驟7。

步驟5：對(duì)數(shù)據(jù)的put事務(wù)性的判斷，是否滿足要求，如果滿足，則轉(zhuǎn)至步驟6，否則，轉(zhuǎn)至步驟5。

步驟6：將數(shù)據(jù)寫入內(nèi)存，并且為下次的寫入做準(zhǔn)備，判斷MemChannel是否為空或者FileChannel現(xiàn)在大小是否超過100，二者只要滿足其一，就將putToMemChannel置為false。

步驟7：對(duì)數(shù)據(jù)進(jìn)行put進(jìn)行事務(wù)性的判斷，是否滿足要求，如果滿足，則將數(shù)據(jù)通過寫入FileChannel，如果不滿，則轉(zhuǎn)至步驟7。

步驟8：執(zhí)行take方法，判斷是從哪個(gè)類型的channel獲取數(shù)據(jù)，如果是MemChannel，則轉(zhuǎn)至步驟9，否則轉(zhuǎn)至步驟11。

步驟9：對(duì)數(shù)據(jù)的take事務(wù)性進(jìn)行判斷，判斷是否滿足要求，如果滿足，則轉(zhuǎn)至步驟10，否則，轉(zhuǎn)至步驟9。

步驟10：對(duì)數(shù)據(jù)進(jìn)行事務(wù)性的取出，并判斷數(shù)據(jù)是否為空，如果為空，代表內(nèi)存中無數(shù)據(jù)，將takeFromMemChannel設(shè)為false，意味著下次將從FileChannel中獲取數(shù)據(jù)。

步驟11：對(duì)數(shù)據(jù)進(jìn)行事務(wù)性的取出，并判斷數(shù)據(jù)是否為空，如果為空，代表文件系統(tǒng)中沒有數(shù)據(jù)，則將takeFromMemChannel和putToMemChannel設(shè)為true，意味著下次將從內(nèi)存中讀取和寫入數(shù)據(jù)。

3? 系統(tǒng)實(shí)現(xiàn)與測(cè)試

在構(gòu)建該系統(tǒng)中，該文基于E5-2667CPU（20處理器）、128內(nèi)存和25T硬盤的服務(wù)器上利用VM-ware 虛擬化5臺(tái)服務(wù)器，利用這5臺(tái)服務(wù)器進(jìn)行綜合評(píng)測(cè)。集群安裝組件配置如表1所示。

該文設(shè)計(jì)的系統(tǒng)模型與傳統(tǒng)的Flume-HDFS模型在同一大小數(shù)據(jù)的分析耗時(shí)對(duì)比見圖2，該系統(tǒng)的傳輸效率以及在數(shù)據(jù)的分析實(shí)時(shí)性要高于原系統(tǒng)。

4? 結(jié)語

該文構(gòu)建了基于改進(jìn)的Flume的實(shí)時(shí)數(shù)據(jù)采集系統(tǒng)，通過實(shí)現(xiàn)數(shù)據(jù)采集框架與復(fù)合Channel選擇技術(shù)相結(jié)合，對(duì)數(shù)據(jù)采集環(huán)節(jié)做出了改進(jìn)。實(shí)驗(yàn)測(cè)試了數(shù)據(jù)采集的效率。由實(shí)驗(yàn)結(jié)果表明，相比于傳統(tǒng)的數(shù)據(jù)

分析系統(tǒng)，該系統(tǒng)在數(shù)據(jù)采集效率和時(shí)性方面有明顯的提升。綜合分析可得，該文平臺(tái)可以更加高效地、穩(wěn)定地完成數(shù)據(jù)的實(shí)時(shí)采集。

參考文獻(xiàn)

[1] Hossam Hakeem.Layered Software Partterns for Data Analysis in Big Data Environment[J].International Journal of Automation and Computing，2017，14（6）：650-660.

[2] 尚凱.企業(yè)數(shù)據(jù)中心數(shù)據(jù)采集與建模[D].山東大學(xué)，2017.

[3] 于秦.基于Apache Flume的大數(shù)據(jù)日志收集系統(tǒng)[J].中國新通信，2016，18（18）：41.

[4] 李洋，呂家恪.基于Hadoop與Storm的日志實(shí)時(shí)處理系統(tǒng)研究[J].西南師范大學(xué)學(xué)報(bào)：自然科學(xué)版，2017，42（4）：119-126.

[5] 陳飛，艾中良.基于Flume的分布式日志采集分析系統(tǒng)設(shè)計(jì)與實(shí)現(xiàn)[J].軟件，2016，37（12）：82-88.

[6] M.Rashid，A.Hamid，N.Ahmad，et al. Novel Machine Learning Approach for Sentiment Analysis of Real Time Twitter Data with Apache Flume[C]//2020 Sixth International Conference on Parallel，Distributed and Grid Computing （PDGC）.2020：336-340.

[7] A.Kanavos，G.Vonitsanos，A.Mohasseb，et al. An Entropy-based Evaluation for Sentiment Analysis of Stock Market Prices using Twitter Data[C]//2020 15th International Workshop on Semantic and Social Media Adaptation and Personalizatio.2020：1-7.