基于Flink的網(wǎng)絡(luò)內(nèi)容分析系統(tǒng)設(shè)計(jì)與實(shí)現(xiàn)＊

趙佳欣，韓 鈺，石明翔，李 碩，白碩

(北京城市學(xué)院信息學(xué)部，北京 100074)

0 引言

在新媒體時(shí)代，各式各樣的網(wǎng)絡(luò)內(nèi)容呈現(xiàn)爆炸式增長(zhǎng)的趨勢(shì)，互聯(lián)網(wǎng)的各類社交媒體平臺(tái)已成為公眾獲取信息、表達(dá)觀點(diǎn)的重要平臺(tái)，人們表達(dá)觀點(diǎn)的同時(shí)也增加了因網(wǎng)絡(luò)輿情內(nèi)容引發(fā)的各種事件。傳統(tǒng)的網(wǎng)絡(luò)分析系統(tǒng)是利用復(fù)雜、高成本的硬件結(jié)合人力來實(shí)現(xiàn)的，存在一定的局限性和不穩(wěn)定性，已經(jīng)不能滿足實(shí)際需求。

隨著我國科學(xué)技術(shù)不斷發(fā)展進(jìn)步，一些學(xué)者也曾嘗試?yán)么髷?shù)據(jù)技術(shù)解決網(wǎng)絡(luò)輿情分析系統(tǒng)的海量數(shù)據(jù)實(shí)時(shí)處理需求，例如諶志華利用Hadoop技術(shù)進(jìn)行數(shù)據(jù)處理，利用HDFS 進(jìn)行數(shù)據(jù)存儲(chǔ)。但是Hadoop 主要是面向靜態(tài)數(shù)據(jù)的批處理，未能很好地解決復(fù)雜高速的實(shí)時(shí)處理和分析的問題[1]。因此，本文嘗試?yán)么髷?shù)據(jù)架構(gòu)Kappa、實(shí)時(shí)計(jì)算框架Flink、數(shù)據(jù)存儲(chǔ)Iceberg 及文本挖掘相關(guān)技術(shù)及算法，實(shí)現(xiàn)《基于Flink的網(wǎng)絡(luò)內(nèi)容分析系統(tǒng)》。通過本系統(tǒng)可更加精準(zhǔn)快速地分析海量網(wǎng)絡(luò)內(nèi)容，幫助有關(guān)部門提高對(duì)網(wǎng)絡(luò)內(nèi)容的監(jiān)管效率，達(dá)到對(duì)網(wǎng)絡(luò)環(huán)境保護(hù)、凈化的目的，從而維護(hù)社會(huì)的和諧安定。

1 系統(tǒng)架構(gòu)及功能流程

本系統(tǒng)使用Kappa 架構(gòu)[2]搭建，系統(tǒng)由應(yīng)用服務(wù)層、數(shù)據(jù)計(jì)算層、數(shù)據(jù)存儲(chǔ)層、數(shù)據(jù)清洗處理層和數(shù)據(jù)采集層構(gòu)成，主要功能包括數(shù)據(jù)采集、數(shù)據(jù)清洗處理、文本挖掘、數(shù)據(jù)計(jì)算、數(shù)據(jù)多維分析及可視化展示，其系統(tǒng)架構(gòu)如圖1所示。

圖1 系統(tǒng)架構(gòu)

首先數(shù)據(jù)采集層會(huì)通過爬蟲的方式獲取數(shù)據(jù)并將其提供給Kafka，數(shù)據(jù)清洗處理層再按照順序依次對(duì)數(shù)據(jù)進(jìn)行數(shù)據(jù)去重、數(shù)據(jù)清洗，之后將完成預(yù)處理的數(shù)據(jù)執(zhí)行文本分詞、去停用詞、關(guān)鍵詞提取等文本挖掘流程，接著數(shù)據(jù)存儲(chǔ)層通過消息隊(duì)列Kafka 將數(shù)據(jù)保存至Iceberg，然后數(shù)據(jù)計(jì)算層利用Flink 并根據(jù)應(yīng)用服務(wù)層實(shí)際需求進(jìn)行實(shí)時(shí)計(jì)算和離線計(jì)算，離線計(jì)算的結(jié)果保存至Hbase。當(dāng)應(yīng)用服務(wù)層發(fā)送請(qǐng)求時(shí)，將實(shí)時(shí)和離線的計(jì)算結(jié)果分別發(fā)送至應(yīng)用服務(wù)層。最后應(yīng)用服務(wù)層將內(nèi)容呈現(xiàn)給用戶。

2 系統(tǒng)實(shí)現(xiàn)

2.1 數(shù)據(jù)采集層

數(shù)據(jù)采集層負(fù)責(zé)以微博網(wǎng)站為數(shù)據(jù)源采集數(shù)據(jù)，數(shù)據(jù)集包括100w條數(shù)據(jù)，數(shù)據(jù)項(xiàng)分別是網(wǎng)絡(luò)內(nèi)容的鏈接、熱搜數(shù)、發(fā)布時(shí)間、轉(zhuǎn)發(fā)數(shù)、評(píng)論數(shù)、點(diǎn)贊數(shù)及發(fā)布者ID、名稱、關(guān)注數(shù)、粉絲數(shù)和歷史發(fā)布內(nèi)容相關(guān)信息等內(nèi)容。

2.2 數(shù)據(jù)清洗處理層

數(shù)據(jù)清洗處理層是利用Flink 的在線流式處理。從Kafka 獲取到網(wǎng)絡(luò)爬蟲數(shù)據(jù)后，首先進(jìn)行數(shù)據(jù)預(yù)處理，然后通過創(chuàng)建前綴字典樹構(gòu)建有向無環(huán)圖的方式再結(jié)合動(dòng)態(tài)規(guī)劃算法得到分詞結(jié)果[3]，經(jīng)過停用詞篩選后利用TF-IDF 算法抽取關(guān)鍵詞，并將其輸入到Kafka中。Flink清洗處理流程如圖2所示。

圖2 Flink清洗處理流程

當(dāng)進(jìn)行中文分詞計(jì)算任務(wù)時(shí)，首先需要建立前綴字典樹，它可根據(jù)字符串的公共前綴來減少查詢時(shí)間，最大限度地減少字符串比較[4]。然后構(gòu)造出有向無環(huán)圖再依據(jù)n-gram 算法計(jì)算出最大概率路徑進(jìn)而得到最終的分詞結(jié)果。在n-gram 算法中需計(jì)算出每句話的概率，設(shè)s 表示句子，由i 個(gè)詞序列w1,w2,w3...wi組成，所以句子的概率可通過公式⑴來計(jì)算，但由于計(jì)算過于復(fù)雜，通常采用公式⑵的方式來計(jì)算

在利用TF-IDF 算法提取關(guān)鍵詞的過程中，由于特定內(nèi)容在篇幅長(zhǎng)的文章里有更高的詞頻，為了防止偏向于篇幅長(zhǎng)的網(wǎng)絡(luò)內(nèi)容，在計(jì)算時(shí)會(huì)將詞頻歸一化[5]?？梢罁?jù)公式⑶、公式⑷、公式⑸來計(jì)算詞頻和逆文檔頻率，進(jìn)而提取出關(guān)鍵詞。其中tfd,t表示詞語t 在語料庫d 中出現(xiàn)的頻率，nd,t表示詞語t 在語料庫d 中出現(xiàn)的次數(shù)，|D|代表語料庫中文件總數(shù)，dft表示包含詞語t的語料庫個(gè)數(shù)。

TF計(jì)算公式：

IDF計(jì)算公式：

2.3 數(shù)據(jù)存儲(chǔ)層

數(shù)據(jù)存儲(chǔ)層負(fù)責(zé)按需向數(shù)據(jù)計(jì)算層提供數(shù)據(jù)。系統(tǒng)中擁有海量的數(shù)據(jù)，需要Kafka 作為消息隊(duì)列進(jìn)行消息的緩沖，保證系統(tǒng)的高效性及穩(wěn)定性[6]。除此之外，系統(tǒng)還包含多種業(yè)務(wù)模塊，數(shù)據(jù)湖Iceberg 作為保存數(shù)據(jù)的中間件，支持隱藏分區(qū)和分區(qū)進(jìn)化的特性，便于數(shù)據(jù)寫入和業(yè)務(wù)數(shù)據(jù)分區(qū)策略更新。Flink 數(shù)據(jù)寫入Iceberg流程如圖3所示。

圖3 Flink數(shù)據(jù)寫入Iceberg流程

Iceberg 擁有ACID 的能力可以降低數(shù)據(jù)入庫到處理流程的延遲問題。它支持流任務(wù)和批任務(wù)使用相同的存儲(chǔ)模型，可更好實(shí)現(xiàn)Flink流批一體的特性。在數(shù)據(jù)計(jì)算層執(zhí)行實(shí)時(shí)計(jì)算任務(wù)時(shí)，Iceberg 還支持流式讀取增量數(shù)據(jù)，可快速讀取到上游數(shù)據(jù)，保證當(dāng)前任務(wù)的正常運(yùn)行[7]。Flink 查詢Iceberg 數(shù)據(jù)流程如圖4所示。

圖4 Flink查詢Iceberg數(shù)據(jù)流程

2.4 數(shù)據(jù)計(jì)算層

數(shù)據(jù)計(jì)算層負(fù)責(zé)將數(shù)據(jù)從Iceberg 中抽取，然后進(jìn)行數(shù)據(jù)分析計(jì)算，將得出的結(jié)果提供給應(yīng)用服務(wù)接口。該層采用Flink進(jìn)行高吞吐低延遲的計(jì)算，因其具有很好的抗壓能力[8]。當(dāng)網(wǎng)絡(luò)爬蟲的數(shù)據(jù)不斷輸送到消息隊(duì)列中時(shí)，F(xiàn)link 可按順序消費(fèi)數(shù)據(jù)。由于Flink流批一體的特性，在具體實(shí)現(xiàn)時(shí)能做到精準(zhǔn)快速地完成任務(wù)。本層的計(jì)算內(nèi)容包括熱點(diǎn)計(jì)算、追蹤計(jì)算、類別計(jì)算、趨勢(shì)計(jì)算及用戶畫像計(jì)算。在進(jìn)行數(shù)據(jù)計(jì)算時(shí)，分為實(shí)時(shí)的流處理計(jì)算和離線的批處理計(jì)算。實(shí)時(shí)流式計(jì)算負(fù)責(zé)統(tǒng)計(jì)本日的數(shù)據(jù)，離線批式計(jì)算負(fù)責(zé)統(tǒng)計(jì)本日之前的所有歷史數(shù)據(jù)，最后，將兩組計(jì)算結(jié)果匯總，返回給應(yīng)用服務(wù)層[9]。具體每部分的Flink算子設(shè)計(jì)和計(jì)算流程如圖5所示。

2.5 應(yīng)用服務(wù)層

應(yīng)用層是系統(tǒng)前端部分，負(fù)責(zé)與用戶進(jìn)行直接交互，可將用戶的請(qǐng)求發(fā)送至應(yīng)用服務(wù)接口，待其處理后返回相應(yīng)結(jié)果并展示。該層包括網(wǎng)絡(luò)內(nèi)容的熱點(diǎn)分析、趨勢(shì)分析、類別分析、追蹤分析和用戶畫像等功能。

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

為了驗(yàn)證系統(tǒng)的執(zhí)行效率，本文集群環(huán)境采用3臺(tái)虛擬機(jī)，一個(gè)為主節(jié)點(diǎn)Master，兩個(gè)子節(jié)點(diǎn)Slave1、Slave2，虛擬機(jī)硬件配置均為16G 內(nèi)存和512G 硬盤，使用的操作系統(tǒng)為Centos7 64 位。在實(shí)驗(yàn)中，利用搭建好的集群處理不同量級(jí)的數(shù)據(jù)，并將其執(zhí)行效率與Hadoop、Spark 執(zhí)行效果進(jìn)行對(duì)比，不同平臺(tái)執(zhí)行效率如圖6所示。

圖6 平臺(tái)執(zhí)行效率

實(shí)驗(yàn)結(jié)果表明，在處理低量級(jí)數(shù)據(jù)時(shí)Hadoop平臺(tái)、Spark 平臺(tái)和Flink 平臺(tái)的執(zhí)行效率沒有明顯的差別。但隨著數(shù)據(jù)量的增加，不同平臺(tái)的執(zhí)行效率逐漸有了顯著差異。Hadoop平臺(tái)和Spark平臺(tái)的響應(yīng)時(shí)間大幅增加導(dǎo)致執(zhí)行效率變低[10]，而Flink 平臺(tái)在處理大量數(shù)據(jù)時(shí)仍能做出快速響應(yīng)。實(shí)驗(yàn)結(jié)果證明本系統(tǒng)的執(zhí)行效率隨著數(shù)據(jù)量的增大，依然能做到快速響應(yīng)，表現(xiàn)出系統(tǒng)良好的執(zhí)行效率。

4 結(jié)束語

本文依托于大數(shù)據(jù)架構(gòu)Kappa，利用實(shí)時(shí)計(jì)算框架Flink、數(shù)據(jù)存儲(chǔ)Iceberg、消息隊(duì)列Kafka 及文本挖掘相關(guān)算法包括n-gram算法和TF-IDF算法，實(shí)現(xiàn)《基于Flink 的網(wǎng)絡(luò)內(nèi)容分析系統(tǒng)》，用于對(duì)網(wǎng)絡(luò)內(nèi)容進(jìn)行實(shí)時(shí)分析。在系統(tǒng)實(shí)現(xiàn)過程中，本文重點(diǎn)解決了實(shí)時(shí)數(shù)據(jù)處理、Flink 流批一體任務(wù)執(zhí)行、實(shí)時(shí)分析網(wǎng)絡(luò)內(nèi)容等技術(shù)難題，并通過實(shí)驗(yàn)結(jié)果的對(duì)比分析證明了系統(tǒng)無論是在運(yùn)行效率還是可擴(kuò)展性上都有著良好的表現(xiàn)。