基于Spark的實(shí)時(shí)數(shù)據(jù)采集與處理

黃 濤 高麗婷

(河北建筑工程學(xué)院,河北 張家口 075000)

0 引 言

大數(shù)據(jù)的時(shí)代背景下，數(shù)據(jù)越來(lái)越重要，如何把數(shù)據(jù)的最大價(jià)值挖掘出來(lái)是一個(gè)不可避免的問(wèn)題.為挖掘數(shù)據(jù)的最大價(jià)值，對(duì)數(shù)據(jù)的分析被分為對(duì)歷史數(shù)據(jù)的分析和實(shí)時(shí)數(shù)據(jù)分析.實(shí)時(shí)數(shù)據(jù)處理應(yīng)用的場(chǎng)景也越來(lái)越多，如文獻(xiàn)[1]分析了實(shí)時(shí)數(shù)據(jù)分析在工業(yè)方面的作用.Spark是實(shí)時(shí)數(shù)據(jù)分析中常用的一種平臺(tái)，它提供流式處理框架用于實(shí)時(shí)數(shù)據(jù)的處理.本文借助Spark搭建完整的大數(shù)據(jù)平臺(tái)來(lái)完成本次實(shí)驗(yàn)分析.本方案以分析學(xué)習(xí)網(wǎng)站用戶喜好為目標(biāo)，主要完成對(duì)網(wǎng)絡(luò)實(shí)時(shí)數(shù)據(jù)進(jìn)行采集與實(shí)時(shí)分析的實(shí)現(xiàn).使用Flume對(duì)網(wǎng)站日志文件進(jìn)行實(shí)時(shí)監(jiān)控與數(shù)據(jù)采集，并將數(shù)據(jù)下發(fā)給Kafka組件.再由Spark streaming對(duì)采集到的信息進(jìn)行過(guò)濾與統(tǒng)計(jì)處理，將處理后的數(shù)據(jù)存入到Hbase中持久化數(shù)據(jù)加以利用.

1 系統(tǒng)結(jié)構(gòu)設(shè)計(jì)

該方法主要包括三個(gè)模塊，分別是Flume模塊、Kafka模塊、Spark streaming流處理模塊.Flume模塊處于系統(tǒng)頂層中，主要負(fù)責(zé)對(duì)文件的監(jiān)控與分發(fā).置于中間層的是Kafka模塊，主要負(fù)責(zé)接收上層的消息并提供一個(gè)給下一層使用的消息隊(duì)列.底層是Spark streaming流處理模塊，主要負(fù)責(zé)實(shí)時(shí)處理消息流數(shù)據(jù)與轉(zhuǎn)存處理好的數(shù)據(jù).通過(guò)三個(gè)模塊互相協(xié)調(diào)，各個(gè)組件之間的消息通信，完成了實(shí)時(shí)日志數(shù)據(jù)的產(chǎn)生到監(jiān)控，最后統(tǒng)計(jì)整理存入到數(shù)據(jù)庫(kù)中加以利用.

1.1 Flume模塊

Flume是一個(gè)用于大數(shù)據(jù)平臺(tái)的文件采集，聚類和傳輸?shù)南到y(tǒng).可作用于分布式系統(tǒng)中.Flume工作是基于Agent的.Agent又包含了Source、Channel、Sink.這三個(gè)組件是Flume的核心組件，表1分別對(duì)其進(jìn)行說(shuō)明.

表1 Flume組件說(shuō)明

1.2 Kafka模塊

Kafka，一種消息發(fā)布與訂閱系統(tǒng).可以將Kafka理解為一種消息隊(duì)列，且特別是用來(lái)處理大數(shù)據(jù)量的一種消息隊(duì)列.Kafka處理消息的核心組件有生產(chǎn)者(producer)，消費(fèi)者(consumer)，主題(topic).Producer，消息的生產(chǎn)者，負(fù)責(zé)源源不斷的產(chǎn)生消息，本文的消息來(lái)源是上一級(jí)的Flume.Topic，消息處理主題，可以理解為一個(gè)隊(duì)列.Producer產(chǎn)生的消息要確定是發(fā)給哪一個(gè)主題來(lái)處理，同樣消費(fèi)者在消費(fèi)時(shí)也要根據(jù)主題來(lái)確定消費(fèi)的是哪里的數(shù)據(jù).Consumer，消息消費(fèi)者，從確定的主題隊(duì)列中得到消息并處理.

Kafka的生產(chǎn)者產(chǎn)生的消息可以被消費(fèi)一次或多次，由此衍生了Kafka的兩種消費(fèi)模式.第一種為一對(duì)一，即消息只會(huì)被消費(fèi)一次，之后會(huì)被刪除.而另外一種為一對(duì)多，即消息被消費(fèi)多次，首次消費(fèi)該消息后，消息不會(huì)被刪除，繼續(xù)保留在消費(fèi)隊(duì)列中，等待再次消費(fèi)，且可以是不同的消費(fèi)者來(lái)消費(fèi).

1.3 Spark streaming流處理模塊

Spark streaming是一款可以使用復(fù)雜算子來(lái)處理流數(shù)據(jù)的實(shí)時(shí)數(shù)據(jù)處理工具，可以使用高級(jí)函數(shù)或者是機(jī)器學(xué)習(xí)算法來(lái)處理實(shí)時(shí)數(shù)據(jù).

Spark streaming工作原理，是使用微批次處理的思想將接收到的實(shí)時(shí)數(shù)據(jù)抽象為Dstream，即離散化數(shù)據(jù)流.其核心本質(zhì)就是將數(shù)據(jù)流根據(jù)時(shí)間節(jié)點(diǎn)的不同來(lái)劃分為很多的批次，對(duì)每個(gè)批次進(jìn)行處理，做出快速響應(yīng)分析來(lái)達(dá)到實(shí)時(shí)處理的效果.使用streaming模塊具有下一優(yōu)勢(shì)，首先spark streaming具有動(dòng)態(tài)負(fù)載均衡的特點(diǎn)，即根據(jù)每個(gè)節(jié)點(diǎn)的特性來(lái)動(dòng)態(tài)的分配任務(wù)，這樣就可以合理利用每個(gè)節(jié)點(diǎn)的計(jì)算資源.其次spark streaming有很高的容錯(cuò)性，streaming作業(yè)會(huì)被拆分為在任何地方都能運(yùn)行，運(yùn)行后合并的結(jié)果不會(huì)產(chǎn)生變化的小任務(wù)，它能夠并行處理，容錯(cuò)性較高.然后由于spark還具有很好的可擴(kuò)展性，每臺(tái)主機(jī)只要經(jīng)過(guò)相應(yīng)的配置就可以加入到指定的spark集群中去，所以在streaming中可以根據(jù)作業(yè)所需的計(jì)算資源動(dòng)態(tài)的調(diào)整集群中的主機(jī)數(shù)量，合理利用每個(gè)計(jì)算資源，減少不必要的資源浪費(fèi).

2 實(shí)時(shí)數(shù)據(jù)集的模擬

2.1 實(shí)時(shí)數(shù)據(jù)背景

本文想要模擬出一個(gè)學(xué)習(xí)網(wǎng)站中課程點(diǎn)擊次數(shù)以及使用哪種搜索引擎的比例.要分析學(xué)習(xí)網(wǎng)站課程點(diǎn)擊與搜索引擎的使用，需要對(duì)其日志進(jìn)行分析，需要提取日志中的關(guān)鍵信息包括IP地址、時(shí)間、請(qǐng)求方式、請(qǐng)求的課程、請(qǐng)求的協(xié)議、狀態(tài)碼及使用的搜索引擎.根據(jù)以上分析，時(shí)間可以使用本地時(shí)間，由于請(qǐng)求方式和協(xié)議對(duì)結(jié)果沒(méi)有影響，默認(rèn)全部為”GET”和”HTTP1.1”.還需要五個(gè)數(shù)組來(lái)定義其他關(guān)鍵信息，然后隨機(jī)組合產(chǎn)生完整的日志信息.

2.2 實(shí)時(shí)數(shù)據(jù)產(chǎn)生

創(chuàng)建以下五個(gè)數(shù)組用來(lái)模擬實(shí)時(shí)數(shù)據(jù)，各數(shù)組的功能與描述如表2所示.

表2 數(shù)組功能與描述

創(chuàng)立目標(biāo)數(shù)組后，對(duì)所有日志關(guān)鍵信息進(jìn)行拼接形成完整的日志信息.一條完整的日志信息格式為：{64.156.29.632022-04-24 13:52:35"GET /class/500.html HTTP/1.1"200http://www.baidu.com/s?wd=Linux進(jìn)階}.本條日志信息表示ip為64.156.29.63的主機(jī)在2022年4月24日的13點(diǎn)52分點(diǎn)擊了課程編號(hào)為500的課程，使用的HTTP請(qǐng)求和GET方式訪問(wèn)，響應(yīng)的狀態(tài)碼為200，并且使用了百度搜索引擎搜索了關(guān)鍵字為L(zhǎng)inux進(jìn)階的網(wǎng)頁(yè)信息.

實(shí)時(shí)效果模擬.為達(dá)到實(shí)時(shí)數(shù)據(jù)模擬的效果.在Linux的終端環(huán)境下，使用命令crontab來(lái)定期執(zhí)行任務(wù)的.設(shè)置定期執(zhí)行的間隔，每隔指定時(shí)間產(chǎn)生運(yùn)行一次腳本來(lái)產(chǎn)生數(shù)據(jù)達(dá)到實(shí)時(shí)數(shù)據(jù)模擬的效果.可以根據(jù)實(shí)際需求來(lái)設(shè)置定時(shí)執(zhí)行的間隔和每次執(zhí)行使產(chǎn)生多少實(shí)時(shí)日志數(shù)據(jù)，部分實(shí)時(shí)數(shù)據(jù)如圖1所示.

圖1 部分實(shí)時(shí)日志數(shù)據(jù)

3 實(shí)驗(yàn)分析

3.1 實(shí)驗(yàn)環(huán)境

由于Spark是基于Hadoop的運(yùn)行的，所以實(shí)驗(yàn)環(huán)境必須搭建Hadoop作為基礎(chǔ)，而Kafka也依賴與Zookeeper運(yùn)行，要搭建Zookeeper保證Kafka的運(yùn)行.在搭建好所有組件后，確保每個(gè)組件的配置文件的修改都生效并可以成功運(yùn)行，想要集群正常運(yùn)行還需配置hosts主機(jī)名與主機(jī)地址的映射，配置SSH免密登錄.本實(shí)驗(yàn)所依賴的運(yùn)行平臺(tái)環(huán)境整體如表3所示.

表3 實(shí)驗(yàn)整體環(huán)境

3.2 實(shí)驗(yàn)結(jié)果

啟動(dòng)整個(gè)實(shí)驗(yàn)環(huán)境，確保環(huán)境中的每個(gè)組件都正常運(yùn)行，且在統(tǒng)一分布式的環(huán)境下.啟動(dòng)一個(gè)flume事件來(lái)監(jiān)控日志文件，當(dāng)日志文件被寫(xiě)入日志時(shí)就會(huì)被監(jiān)控.之后創(chuàng)建一個(gè)Kafka的topic來(lái)接收所監(jiān)控的日志文件.之后啟動(dòng)Spark streaming實(shí)時(shí)處理Kafka所得到的日志，對(duì)其篩選，統(tǒng)計(jì)等操作之后存入Hbase數(shù)據(jù)庫(kù)令其持久化，以便后續(xù)分析與操作.在Linux的終端中周期性的運(yùn)行日志產(chǎn)生的腳本，flume會(huì)實(shí)時(shí)監(jiān)控日志是否被產(chǎn)生新的日志，日志文件發(fā)生變化，flume就會(huì)給Kafka新的數(shù)據(jù)流處理，就可以達(dá)到預(yù)期的實(shí)時(shí)處理的效果，成功對(duì)實(shí)時(shí)產(chǎn)生的日志進(jìn)行了實(shí)時(shí)處理并存儲(chǔ)到了Hbase中.實(shí)時(shí)數(shù)據(jù)分析效果如圖2、圖3所示.

圖2 搜索引擎統(tǒng)計(jì)表

圖3 課程統(tǒng)計(jì)表

在圖2中，左邊是搜索引擎的列表，由時(shí)間和搜索引擎的名字組成，右邊是時(shí)間戳和具體值，某一行的內(nèi)容表示某搜索引擎的使用次數(shù).通過(guò)分析對(duì)應(yīng)搜索引擎標(biāo)簽的具體點(diǎn)擊次數(shù)就可以得知哪種搜索引擎是使用最多次數(shù)的，哪種是最不喜歡的，對(duì)應(yīng)著使用次數(shù)最少.達(dá)到分析搜索引擎喜好的目標(biāo).

在圖3中，左邊的一列是時(shí)間加上課程的編號(hào)，右邊同樣是具體時(shí)間戳和次數(shù)的值，某一行的內(nèi)容表示該課程共被點(diǎn)擊了多少次.根據(jù)課程編號(hào)還原對(duì)應(yīng)的課程，在結(jié)合具體的使用數(shù)值，可以清楚的分析出哪種課程學(xué)習(xí)人數(shù)最多，最受喜愛(ài)，哪種課程學(xué)習(xí)人數(shù)最少，不受歡迎.達(dá)到分析出課程的受歡迎程度.

4 結(jié) 論

本文從環(huán)境搭建，到分析實(shí)時(shí)日志，做了完整的實(shí)驗(yàn).編寫(xiě)了腳本實(shí)時(shí)產(chǎn)生日志數(shù)據(jù)文件，使用了Flume對(duì)文件進(jìn)行實(shí)時(shí)監(jiān)控，交由Kafka進(jìn)行消息整合與分發(fā)，最后交付給spark streaming處理，發(fā)現(xiàn)Spark在實(shí)時(shí)數(shù)據(jù)處理中具有很高的運(yùn)算能力，它的streaming流處理框架特別適用于實(shí)時(shí)數(shù)據(jù)的處理.如今的環(huán)境下，數(shù)據(jù)量越來(lái)越高，正是需要高處理能力的大數(shù)據(jù)處理平臺(tái).數(shù)據(jù)的實(shí)時(shí)分析也越來(lái)越重要，在實(shí)時(shí)分析中還可以處理異常數(shù)據(jù)，并將處理的數(shù)據(jù)轉(zhuǎn)存到其他的數(shù)據(jù)庫(kù)中持久化使用.