朱奕健 張正卿

【摘要】 本文利用流式數(shù)據(jù)處理框架探索了一種新的基于運營商實時大數(shù)據(jù)業(yè)務(wù)系統(tǒng)構(gòu)建模式。首先，在充分研究了業(yè)內(nèi)實時流式處理技術(shù)的發(fā)展以及運營商本身實時數(shù)據(jù)源的特點之后，確定以Flume作為實時采集和分流組件，Kafka作為緩存和多模塊通信組件，以Spark Streaming的分布式結(jié)構(gòu)作為數(shù)據(jù)ETL集群；然后，利用該系統(tǒng)進行了重點區(qū)域的人流實時監(jiān)控的業(yè)務(wù)，在實施過程中為了提供毫秒級的數(shù)據(jù)結(jié)果流查詢能力，采用了Redis組件提供基于內(nèi)存的Key-Value引擎；最終，通過流式數(shù)據(jù)處理效率的對比和實時監(jiān)控的人流效果，我們驗證了了這種新的技術(shù)架構(gòu)針對運營商CS域和PS域數(shù)據(jù)實時處理需求的可行性，結(jié)果表明，新的實時業(yè)務(wù)架構(gòu)能更加有效的提高從實時采集到業(yè)務(wù)觸發(fā)的運行效率，并且為公安部門在重大節(jié)假日的區(qū)域級人流監(jiān)控、預(yù)警和疏導(dǎo)提供了技術(shù)保障。

【關(guān)鍵詞】 大數(shù)據(jù) 流處理 簡單事件處理引擎（PME） Flume Kafka

一、引言

隨著網(wǎng)絡(luò)、通信和傳感器應(yīng)用的飛速發(fā)展，尤其是移動通信全面進入移動互聯(lián)網(wǎng)時代，直接帶來通信網(wǎng)絡(luò)中的數(shù)據(jù)復(fù)雜度、信息量迅速增長，諸多的移動設(shè)備實時收集用戶各種信息，如位置、喜愛偏好、移動軌跡、血壓、體溫等，帶來數(shù)據(jù)的規(guī)模、種類和關(guān)聯(lián)性等急劇膨脹。“大數(shù)據(jù)”成為時下各個行業(yè)中出現(xiàn)頻率最高的關(guān)鍵詞之一。思科估算在2015年僅移動網(wǎng)絡(luò)的數(shù)據(jù)量將突破6EB/月，相當于億字節(jié)的海量數(shù)據(jù)；而IDC預(yù)計到2020年全世界的數(shù)據(jù)存儲總量將達到35萬億GB。大數(shù)據(jù)時代的到來使得隱藏在海量數(shù)據(jù)中的信息開始深刻的影響著人們的日常生活。當顧客在網(wǎng)上購物時，推薦系統(tǒng)會根據(jù)從海量數(shù)據(jù)中挖掘出的信息向其推薦適合的商品；當乘客出行時，打車軟件又替他們搜索周圍空閑出租車并選擇最優(yōu)車輛來提供服務(wù)；當病人看病時，醫(yī)生又會根據(jù)該病人的日常醫(yī)療數(shù)據(jù)制定最優(yōu)的治療方案。

而隨著4G時代的到來，移動通信業(yè)務(wù)已經(jīng)正式全面進入移動互聯(lián)網(wǎng)時代，飛速發(fā)展的移動網(wǎng)絡(luò)帶寬直接帶來繁雜的應(yīng)用和用戶行為，而通信網(wǎng)絡(luò)中的數(shù)據(jù)復(fù)雜度、信息量都隨之迅速增長，通信運營商所能掌握的數(shù)據(jù)量級與日俱增，導(dǎo)致數(shù)據(jù)處理的復(fù)雜度和運算量要求都隨之有了更高的要求，傳統(tǒng)數(shù)據(jù)庫體系的數(shù)據(jù)處理能力受到了極大的挑戰(zhàn)，面對海量數(shù)據(jù)處理需求和更低的時延性限制要求，傳統(tǒng)數(shù)據(jù)系統(tǒng)投入的CPU計算能力、內(nèi)存響應(yīng)和吞吐、網(wǎng)絡(luò)帶寬都有著巨大的基準，且在高安全性，多中心的發(fā)展趨勢下面臨諸多的瓶頸。

大數(shù)據(jù)時代的到來使單節(jié)點的計算模式已經(jīng)不能滿足數(shù)據(jù)處理的需求，分布式數(shù)據(jù)處理與存儲系統(tǒng)逐步成為大數(shù)據(jù)平臺首選的架構(gòu)，包括Hadoop，MongoDB等開放型的大數(shù)據(jù)技術(shù)成為了眾相研究的熱點。而Hadoop大數(shù)據(jù)平臺主要基于靜態(tài)數(shù)據(jù)文件的并行處理，雖然在海量數(shù)據(jù)吞吐、計算、存儲方面有著極高的效率，但是實時性較差，屬于高吞吐，高并發(fā)，高時延的架構(gòu)，對于小文件的處理性能一直是其不可回避的問題，故針對一些實時性較高的數(shù)據(jù)處理和使用場景下無能為力?；谶@樣的原因，面對動態(tài)數(shù)據(jù)處理的需求，實時流式數(shù)據(jù)處理技術(shù)應(yīng)運而生。

隨著針對數(shù)據(jù)流的研究逐漸進入學術(shù)界，大規(guī)模動態(tài)數(shù)據(jù)集（也稱為實時數(shù)據(jù)流）成為研究及工程人員爭相探索的熱點領(lǐng)域[12]。而實時流式數(shù)據(jù)具有海量性、實時性和動態(tài)變化性三個基本特點，基于這些特性，數(shù)據(jù)研究領(lǐng)域內(nèi)發(fā)展了諸多的研究方向。如流式數(shù)據(jù)處理的數(shù)學工具研究[11]，研究如何保證在數(shù)據(jù)流處理過程中的QoS服務(wù)質(zhì)量[2]，研究利用滑動窗口來實現(xiàn)實時流數(shù)據(jù)處理[1][8]，基于實時性流數(shù)據(jù)查詢算法的優(yōu)化[3]，研究數(shù)據(jù)流的分布式處理和最后聚集[6]，流式數(shù)據(jù)的實時分類[9]。也有融合流處理技術(shù)在其他科技領(lǐng)域來完成復(fù)雜性的計算，如射頻標簽領(lǐng)域的實時數(shù)據(jù)處理[4]，高速網(wǎng)絡(luò)中的數(shù)據(jù)流模型設(shè)計[7]，數(shù)據(jù)流量變化的處理模型[10]。而在大數(shù)據(jù)應(yīng)用領(lǐng)域，更多的企業(yè)在開發(fā)如何利用流處理技術(shù)來構(gòu)造一個企業(yè)級的實時性數(shù)據(jù)業(yè)務(wù)平臺[5]

本論文所有的研究都集中在如何構(gòu)造基于運營商大數(shù)據(jù)流處理系統(tǒng)方面，主要圍繞實時性的業(yè)務(wù)場景下，如何從數(shù)據(jù)產(chǎn)生，數(shù)據(jù)采集，到數(shù)據(jù)流的處理，再到實時業(yè)務(wù)規(guī)則匹配的過程中尋找最佳流式數(shù)據(jù)平臺的架構(gòu)展開研究。全文采用總分結(jié)構(gòu)給出了實時流處理系統(tǒng)的構(gòu)建思路：在第二節(jié)，對實時流處理系統(tǒng)的整體架構(gòu)進行整體性闡述；第三節(jié)主要闡述采用Flume+Kafka+SparkStreaming架構(gòu)來有效解決Hadoop系統(tǒng)對于小數(shù)據(jù)的流式處理效率的提高；在第四節(jié)中，通過該系統(tǒng)成功實現(xiàn)針對固定區(qū)域進行實時人流監(jiān)控的業(yè)務(wù)場景，最后，針對整個系統(tǒng)對于流式處理的效率和實時監(jiān)控效果進行總結(jié)，并形成研究結(jié)論和下一步的研究計劃。

二、流式大數(shù)據(jù)系統(tǒng)綜述

流式大數(shù)據(jù)系統(tǒng)的總體架構(gòu)如圖1所示。

整個系統(tǒng)的流處理框架使用了學術(shù)界內(nèi)公認較為高效的開源組件，整體系統(tǒng)實時數(shù)據(jù)來源于兩方面：

PS域數(shù)據(jù)：基于移動網(wǎng)絡(luò)Gn口中的全量用戶移動上網(wǎng)數(shù)據(jù)；

CS域數(shù)據(jù)：基于信令網(wǎng)絡(luò)中A口中的基站定位數(shù)據(jù)。

在系統(tǒng)底層，利用Flume組件來實時采集匯攏2種來源的數(shù)據(jù)，并根據(jù)上層數(shù)據(jù)需求進行分流，對于需要實時處理的數(shù)據(jù)實時傳送至Kafka集群，對于非實時性數(shù)據(jù)挖掘模型需要的靜態(tài)數(shù)據(jù)來形成文件寫入Hadoop集群的HDFS文件。

在實時數(shù)據(jù)的ETL層，采用Kafka組件來完成所有的數(shù)據(jù)流緩存，該架構(gòu)可以保證整體數(shù)據(jù)流通訊的可靠性以及短時延的對外服務(wù)能力。一些復(fù)雜的數(shù)據(jù)挖掘和處理通過分布式的流式處理結(jié)構(gòu)-Spark streaming來實施，該架構(gòu)充分的結(jié)合了Hadoop的分布式處理的思想和內(nèi)存數(shù)據(jù)庫的實時處理效率，充分保證整體處理過程的高并發(fā)、低時延，并實現(xiàn)了數(shù)據(jù)內(nèi)容和挖掘能力的高魯棒性。而諸多簡單業(yè)務(wù)規(guī)則和數(shù)據(jù)的匹配則通過引入簡單事件匹配引擎（PME）來完成。處理結(jié)果最終會回寫到Kafka中供應(yīng)用層調(diào)用。

在實時業(yè)務(wù)層，實時數(shù)據(jù)處理結(jié)果通過Kafka來被業(yè)務(wù)觸發(fā)系統(tǒng)調(diào)用，結(jié)合Hadoop的靜態(tài)數(shù)據(jù)挖掘結(jié)果（非實時），來形成最終的業(yè)務(wù)觸發(fā)，而在部分業(yè)務(wù)場景需求中，為了提高整個處理效率，我們采用了Redis（內(nèi)存Key-Value引擎）來提供數(shù)據(jù)關(guān)聯(lián)查詢。

三、流式大數(shù)據(jù)處理效率提高

3.1針對靜態(tài)數(shù)據(jù)的小文件處理效率提升

小文件指的是那些Size比HDFS的Block Size（默認64M）小很多的文件。如果在HDFS中存儲許許多多這樣的小文件，我們發(fā)現(xiàn)HDFS根本無法很有效的處理數(shù)量龐大的小文件。任何一個文件，目錄和Block，在HDFS中都會被表示為一個Object存儲在NameNode的內(nèi)存中，每一個Object占用150Bytes的內(nèi)存空間。所以，如果有10 Million個文件，每一個文件對應(yīng)一個Block，那么就將要消耗NameNode 3G的內(nèi)存來保存這些Block的信息。如果規(guī)模再無限制的擴大下去，那么將會超出現(xiàn)階段計算機硬件所能滿足的極限。

不僅如此，HDFS并不是為了有效的處理大量小文件而存在的。它主要是為了流式的訪問大文件而設(shè)計的。對小文件的讀取通常會造成大量從DataNode到DataNode的Seeks和Hopping來Retrieve文件，而這樣是非常的低效的一種訪問方式。

針對Gn口和A口中的實時數(shù)據(jù)結(jié)構(gòu)中存在小容量數(shù)據(jù)塊，如果利用傳統(tǒng)Hadoop結(jié)構(gòu)勢必需要針對數(shù)量巨大的小文件進行高效處理，雖說Hadoop開源組件對小文件提供了許多的解決方案，但是帶來的系統(tǒng)構(gòu)造成本巨大，而且在運營商開展的具體業(yè)務(wù)場景下并不能完全適用。

在利用實時性通訊組件Flume開始接管A口信令數(shù)據(jù)以及Gn口數(shù)據(jù)的采集時，改善平臺小文件狀況的契機開始顯現(xiàn)。我們在Flume中使用了Hadoop一個官方API使之在接收流式數(shù)據(jù)并寫入HDFS時進行文件追加，通過接收一條數(shù)據(jù)追加一條數(shù)據(jù)至當天文件內(nèi)，這種快速積聚小文件到標準大小文件的方式解決了小文件在Hadoop集群中需要較多時延來存儲至HDFS文件的問題。

使用append函數(shù)需要如下兩個步驟：

配置集群（hdfs-site. xml）

dfs.support.append

true

API實現(xiàn)（Flume中數(shù)據(jù)輸出邏輯需要進行API調(diào)用）

String hdfs_path=”hdfs：//ip：xx/file/fileuploadFileName”；//文件路徑

Configuration conf= new Configuration（）；

FileSystem fs= FileSystem.get（URI.create（hdfs_path），conf）；

InputStream in=new BufferedInputStream（new FileInputsStream（file））；

//要追加的文件流，file為文件

IOUtils.copyBytes（in，out，4096，true）；

3.2針對實時性的流數(shù)據(jù)挖掘結(jié)果的結(jié)果查詢技術(shù)

為了保證事件處理結(jié)果的實時讀取，本文選擇Redis來進行結(jié)果存儲。Redis是一個開源、先進的key-value內(nèi)存存儲，用于構(gòu)建高性能、可擴展的Web應(yīng)用程序的完美解決方案。

Redis從它的許多競爭者繼承來的三個主要特點：數(shù)據(jù)庫完全在內(nèi)存中，使用磁盤僅用于持久性；相比許多鍵值數(shù)據(jù)存儲，Redis擁有一套較為豐富的數(shù)據(jù)類型；Redis可以將數(shù)據(jù)復(fù)制到任意數(shù)量的從服務(wù)器。

而對于時延要求極低的結(jié)果查詢使用Redis優(yōu)勢包括：

異?？焖伲篟edis的速度非?？?，每秒能執(zhí)行約11萬集合，每秒約81000+條記錄。

支持豐富的數(shù)據(jù)類型：Redis支持最大多數(shù)開發(fā)人員已經(jīng)知道像列表，集合，有序集合，散列數(shù)據(jù)類型。這使得它非常容易解決各種各樣的問題，因為我們知道哪些問題是可以處理通過它的數(shù)據(jù)類型更好。

操作都是原子性：所有Redis操作是原子的，這保證了如果兩個客戶端同時訪問的Redis服務(wù)器將獲得更新后的值。

多功能實用工具：Redis是一個多實用的工具，可以在多個用例如緩存，消息，隊列使用（Redis原生支持發(fā)布/訂閱），任何短暫的數(shù)據(jù)，應(yīng)用程序，如Web應(yīng)用程序會話，網(wǎng)頁命中計數(shù)等。

整體優(yōu)化前數(shù)據(jù)流程如下：

話單文件→精細化預(yù)處理→HDFS→Hive→GreenPlum

整體優(yōu)化后數(shù)據(jù)流程如下：

話單流→HDFS→Hive→GreenPlum

總處理所需時間較原流程縮短至1/3，效率提高了200%。

四、 流式大數(shù)據(jù)系統(tǒng)實現(xiàn)的業(yè)務(wù)場景

在互聯(lián)網(wǎng)界，百度、亞馬遜、阿里巴巴、京東、騰訊等大型互聯(lián)網(wǎng)企業(yè)將大數(shù)據(jù)的應(yīng)用提高到前所未有的高度，并形成了了一系列滿足各種業(yè)務(wù)需求的大數(shù)據(jù)處理平臺，通過挖掘海量數(shù)據(jù)中蘊含的信息點，并用業(yè)務(wù)流程來關(guān)聯(lián)起來，真正形成數(shù)據(jù)生產(chǎn)力來提高業(yè)務(wù)感知和質(zhì)量，向日益增長的移動互聯(lián)網(wǎng)用戶提供更加優(yōu)質(zhì)的服務(wù)。比如：百度通過典型數(shù)學計算工具結(jié)合Hadoop框架向用戶提供搜索引擎，通過毫秒級DSP處理引擎向廣告服務(wù)提供商實時提供廣告點擊信息；騰訊通過Storm數(shù)據(jù)流處理系統(tǒng)進行簡單的數(shù)據(jù)處理，如過濾、聚類等，以及復(fù)雜數(shù)據(jù)處理，如運行簡單的機器學習算法等；阿里巴巴通過、等計算框架向用戶提供商品的推薦服務(wù)；京東通過海量數(shù)據(jù)的挖掘進行電子商務(wù)的倉儲備貨策略和物流控制策略。在科技學術(shù)界，《自然》雜志于年出版了大數(shù)據(jù)?？?。

2012年10月，哈佛商業(yè)評論上發(fā)表了一篇里程碑式的專題文章《Data scientist： The sexist job of 21st century》，標志著“數(shù)據(jù)科學家”已經(jīng)正式在企業(yè)中收到廣泛的尊重，這類專家的主要工作是從海量非結(jié)構(gòu)化數(shù)據(jù)中挖掘出有價值的信息，而不斷涌現(xiàn)的大數(shù)據(jù)技術(shù)使得從海量數(shù)據(jù)中獲取有價值信息成為可能。

本次實時系統(tǒng)研究過程中，恰逢2015年上海外灘踩踏事故的發(fā)生，考慮到該系統(tǒng)可以基于基站定位實時統(tǒng)計指定區(qū)域內(nèi)的人流量，因此，在與公安系統(tǒng)對接后可以通過有效的技術(shù)手段來預(yù)先判斷人流擁擠程度，避免踩踏事故的再次發(fā)生。

實時人流監(jiān)控架構(gòu)見圖2。

數(shù)據(jù)流程中主要由PME負責訂閱kafka中的簡單的事件并進行處理。將A口信令中Cell_id、Lac字段匹配公參表中的經(jīng)緯度信息，輸出用戶號碼目前匹配的經(jīng)緯度信息。Storm組件負責訂閱復(fù)雜事件并進行處理（如分析小區(qū)用戶群），同時將簡單事件復(fù)雜事件處理結(jié)果輸出至存Redis，以便應(yīng)用頁面能夠快速查詢結(jié)果。

實時人流監(jiān)控系統(tǒng)能力如下（系統(tǒng)界面見圖3）：

客戶信令觸發(fā)30秒后，系統(tǒng)就會捕捉到信令事件，通過2-3秒的計算后即可將用戶位置信息存儲至Redis里。

展示頁面每5秒刷新一次。這樣在頁面內(nèi)展示的內(nèi)容都是1分鐘內(nèi)人流變化情況。

通過在大數(shù)據(jù)平臺引入實時性Flume+Redis的架構(gòu)實現(xiàn)了實時性數(shù)據(jù)采集、處理、展現(xiàn)的大數(shù)據(jù)能力，并利用該能力搭建了從A口信令觸發(fā)開始到最終監(jiān)控界面的幾個熱點區(qū)域?qū)崟r性人流監(jiān)控，該技術(shù)方案在整個運營商業(yè)內(nèi)屬于首創(chuàng)。

五、結(jié)論

本文提出了一種基于運營商網(wǎng)絡(luò)的實時大數(shù)據(jù)系統(tǒng)的構(gòu)建，并成功的利用基于基站定位的實時人流監(jiān)控業(yè)務(wù)來驗證了這種技術(shù)架構(gòu)的合理性，這種模式不僅僅為未來運營商的實時大數(shù)據(jù)業(yè)務(wù)開發(fā)提供了新的思路，同時確保了該技術(shù)架構(gòu)對于具體運營商對外合作業(yè)務(wù)的可實施型。本文下一步工作首先要對實時處理時效繼續(xù)優(yōu)化，構(gòu)建在1分鐘內(nèi)從事件觸發(fā)，源數(shù)據(jù)采集，流處理，到業(yè)務(wù)觸發(fā)到用戶的全流程。然后，增強整個流式數(shù)據(jù)模型開發(fā)，以優(yōu)化數(shù)據(jù)流ETL過程，實現(xiàn)多條實時流大數(shù)據(jù)業(yè)務(wù)的并發(fā)。此外，本文未來的研究工作還將在將在完善實時流處理和運營商推薦平臺融合建設(shè)等方面繼續(xù)開展。

參 考 文 獻

[1] 李俊奎； 王元珍，可重寫循環(huán)滑動窗口：面向高效的在線數(shù)據(jù)流處理[J]. 計算機科學， 2007

[2] 武珊珊，于戈，呂雁飛，谷峪，李曉靜， 數(shù)據(jù)流處理中確定性QoS的保證方法[J]， 軟件學報， 2008

[3] 左利云，馬英杰， 基于數(shù)據(jù)流處理模型的多查詢優(yōu)化算法[J]， 計算機工程與科學， 2009

[4] 陰曉加，鞠時光，王英杰， 基于復(fù)雜事件處理機制的RFID數(shù)據(jù)流處理方法[J]， 計算機應(yīng)用， 2009

[5] 楊瑋，企業(yè)級數(shù)據(jù)中心數(shù)據(jù)流處理方案設(shè)計[J]， 中國科技信息 2007

[6] 侯燕，王永利， 基于近似等深柱狀圖的數(shù)據(jù)流并行聚集算法[J]， 解放軍理工大學學報（自然科學版）， 2008

[7] 陳磊松，林錦賢， 面向高速網(wǎng)絡(luò)的數(shù)據(jù)流處理模型研究[J]， 漳州師范學院學報（自然科學版）， 2006

[8] Yishan Jiao， Maintaining stream statistics over multiscale sliding windows[J]. ACM Transactions on Database Systems （TODS） ， 2006 （4）[9] 徐花芬，毛國君，吳靜， 分布式數(shù)據(jù)流分類關(guān)鍵技術(shù)研究[J]， 華北科技學院學報， 2015

[10] 秦首科，錢衛(wèi)寧，周傲英，基于分形技術(shù)的數(shù)據(jù)流突變檢測算法[J]， ， 2006， （9）

[11] Sudipto Guha， Nick Koudas， Kyuseok Shim， Approximation and streaming algorithms for histogram construction problems[J]， ACM Transactions on Database Systems （TODS）， 2006 （1）

[12] 張鵬，李鵬霄，任彥，林海倫，楊嶸，鄭超，面向大數(shù)據(jù)的分布式流處理技術(shù)綜述，計算機研究與發(fā)展[J]，2014