李云鵬

（西北工業(yè)大學 陜西 西安 710129）

0 引言

隨著電子商務、移動網(wǎng)絡以及社交軟件的興起，將我們推向了一個以PB（1024TB）為單位擁有海量信息的大數(shù)據(jù)時代。面對海量的數(shù)據(jù)信息，我們迫切需要以更高效的方法進行數(shù)據(jù)挖掘、數(shù)據(jù)處理，這也是我們在大數(shù)據(jù)時代所面臨的新挑戰(zhàn)。為解決數(shù)據(jù)的存儲和計算方面的難題，需要大數(shù)據(jù)技術來構建大數(shù)據(jù)平臺[1]。傳統(tǒng)的大數(shù)據(jù)平臺包括Hadoop、Spark、Flink等。這些大數(shù)據(jù)平臺具有對海量數(shù)據(jù)進行分析的能力，可以將海量數(shù)據(jù)轉(zhuǎn)化為生產(chǎn)力，從而產(chǎn)生實際價值。針對不同的數(shù)據(jù)類型以及復雜的應用場景，對數(shù)據(jù)的計算模式分為以下四種：批處理計算、流計算、圖計算以及查詢分析計算。本文將聚焦于流計算進行探討。

在傳統(tǒng)的數(shù)據(jù)處理過程中，我們先將數(shù)據(jù)存入數(shù)據(jù)庫中，當需要時再去數(shù)據(jù)庫中進行檢索，將處理結(jié)果返回給請求的用戶，更多應用于離線計算場景中。而針對實時性要求較高的場景，我們期望延時在秒甚至是毫秒級別，于是引出了一種新的數(shù)據(jù)計算結(jié)構——流式計算，即對無邊界的數(shù)據(jù)進行連續(xù)不斷地處理。當今市面上主流的Spark平臺，也推出了流式處理子框架——Spark Streaming。Spark Streaming是一個高吞吐、高容錯、低延時的實時處理系統(tǒng)，可以從Kafka、flume、kinesis或者TCP套接字等多種數(shù)據(jù)源中獲取數(shù)據(jù)，然后利用復雜的操作（如map、reduce、window等）對數(shù)據(jù)進行處理，最終將處理后的數(shù)據(jù)輸出到文件系統(tǒng)、數(shù)據(jù)庫或者控制臺上，具體的輸入輸出過程如圖1所示。本文基于流計算的算法，提出一種針對電商信息進行實時展示的系統(tǒng)設計與實現(xiàn)方案。

圖1 Spark Streaming框架的數(shù)據(jù)處理示意圖

1 環(huán)境部署

首先，為高效構建分布式處理系統(tǒng)，從而保證數(shù)據(jù)的高效處理，需在不同的數(shù)據(jù)節(jié)點分別部署好基于Linux（如Ubuntu、Debian等）系統(tǒng)運行環(huán)境，以保證對海量實時數(shù)據(jù)能高效進行模擬分析。本文推薦的大數(shù)據(jù)框架、數(shù)據(jù)庫系統(tǒng)、輔助中間件軟件及版本如下：Hadoop2.7.1，Spark2.4.0，Kafka2.11，Zookeeper3.6.3，MYSQL5.7.31。配置好基本的環(huán)境后，啟動集群環(huán)境，啟動了Zookeeper服務以及Kafka服務后，使用JPS命令進行驗證，如圖2所示。出現(xiàn)Master節(jié)點以及NameNode節(jié)點，證明服務已經(jīng)成功啟動，系統(tǒng)具備可運行條件。

圖2 驗證服務啟動

2 核心技術概述

2.1 Spark Streaming流計算處理框架

與其他大數(shù)據(jù)框架Storm、Flink一樣，Spark Streaming是在Spark Core（一種基于RDD數(shù)據(jù)抽象，用于數(shù)據(jù)并行處理的基礎組件）基礎之上用于處理實時計算業(yè)務的框架[2]。其實現(xiàn)原理就是把輸入的流數(shù)據(jù)按時間切分，切分的數(shù)據(jù)塊用離線批處理的方式進行并行計算。輸入的數(shù)據(jù)流經(jīng)過Spark Streaming的receiver組件，數(shù)據(jù)被切分為DStream，然后DStream被Spark Core的離線計算引擎執(zhí)行并行計算。Spark Streaming與其他主流框架的對比如表1所示。

表1 Spark Streaming與其他主流框架的主要性能對比

2.2 Kafka分布式發(fā)布訂閱消息體系

Kafka是一種高吞吐量的分布式發(fā)布訂閱消息系統(tǒng)，以生產(chǎn)者/消費者模式，通過隊列交換數(shù)據(jù)，在實時計算領域有著非常強大的功能。本系統(tǒng)主要利用Spark Streaming計算框架實時地讀取Kafka中的電商數(shù)據(jù)然后進行并行計算。Kafka在本系統(tǒng)中起到了一個中間件的作用，即搭建了程序間的數(shù)據(jù)管道，用來轉(zhuǎn)換并響應實時數(shù)據(jù)[3]。

2.3 WebSocket通信機制

WebSocket是HTML5中服務端和客戶端進行雙向文本或二進制數(shù)據(jù)通信的一種新協(xié)議。它與HTTP通信協(xié)議不同的是，WebSocket提供全雙工通信[4]。對于傳統(tǒng)的HTTP通信方式，只有當客戶端發(fā)起請求后，服務器端才會發(fā)送數(shù)據(jù)。而WebSocket則可以讓服務器主動發(fā)送數(shù)據(jù)給客戶端，是服務器推送技術的一種[5]。由于本系統(tǒng)后臺產(chǎn)生新的數(shù)據(jù)，需要在前臺頁面實時展示出來，故選擇了WebSocket通信機制。

本系統(tǒng)借助的Flask-socketio模塊，封裝了Flask對WebSocket的支持，WebSocket在連接建立階段是通過HTTP的握手方式進行的，當連接建立后，客戶端和服務端之間就不再進行HTTP通信了，所有信息交互都由WebSocket接管。Flask-SocketIO使Flask應用程序可以訪問客戶端和服務器之間的低延遲雙向通信，使客戶端建立與服務器的永久連接。

2.4 基于Scrapy的爬蟲框架

Scrapy是由Python基于twisted框架開發(fā)的一種高效、快速的web信息抓取框架，用于抓取web站點并從頁面中提取結(jié)構化的數(shù)據(jù)，只需要少量的代碼，就能高效地獲取所需數(shù)據(jù)。Scrapy集成了高性能異步下載、隊列、分布式、解析、持久化等諸多功能。值得注意的是，由于網(wǎng)站數(shù)據(jù)具有一定的商業(yè)價值，需科學合理地使用爬蟲工具進行數(shù)據(jù)獲取，本系統(tǒng)所獲取的數(shù)據(jù)僅用于學術探索，且已經(jīng)獲取相關網(wǎng)站授權。

3 電商信息實時展示系統(tǒng)詳細設計

網(wǎng)上購物，作為一種依托于互聯(lián)網(wǎng)的新型購物方式，為人們的生活帶來了不少的便利。諸如淘寶、京東、亞馬遜等眾多電商平臺產(chǎn)生了大量的購物數(shù)據(jù)，通過合法的爬蟲技術實時爬取這些數(shù)據(jù)，并進行可視化分析，從數(shù)據(jù)中提取關鍵信息，從而有利于對電商平臺進行升級。因此，基于大數(shù)據(jù)相關技術（Spark Streaming+Kafka+We bsocket+Scrapy）本文設計并開發(fā)了電商信息實時展示系統(tǒng)，后期通過源碼剖析，針對并行化的方向進行了系統(tǒng)的優(yōu)化升級。

3.1 總體設計

在本系統(tǒng)中，通過爬取到的電商平臺的交易數(shù)據(jù)及購物日志（注：此類信息為電商平臺主動公布，僅作學術研究，不涉及版權問題；如個別電商平臺未公布，需與平臺進行聯(lián)系，以免侵權），模擬實時數(shù)據(jù)流的產(chǎn)生，用Spark Streaming框架分析每秒購物人數(shù)，每秒鐘平臺上用戶的操作類型等，并利用WebSocket將數(shù)據(jù)實時推送給客戶端，最后瀏覽器將接收到的數(shù)據(jù)實時展現(xiàn)在web端，進行了數(shù)據(jù)的可視化展示。本系統(tǒng)采用模塊化設計思路，共分為以下四個模塊：數(shù)據(jù)獲取模塊、實時數(shù)據(jù)模擬模塊、Spark Streaming實時處理模塊以及可視化展示模塊。本系統(tǒng)總體架構圖如圖3所示。

圖3 系統(tǒng)總體架構圖

3.2 數(shù)據(jù)獲取模塊設計

數(shù)據(jù)獲取模塊通過爬蟲技術，借助Python提供的Scrapy、Requests等庫，爬取電商平臺的購物日志，并以csv的格式存儲到了本地。由于僅做研究使用，本系統(tǒng)中需要關注的數(shù)據(jù)只有：gender項，其中0表示女性，1表示男性，2和NULL表示未知性別；action項，其中0表示點擊行為，1表示加入購物車，2表示購買，3表示關注商品；cat_id項，表示商品類別id。該數(shù)據(jù)集包含上萬條用戶購物日志，可模擬大批量的實時數(shù)據(jù)。然后實時數(shù)據(jù)模擬模塊把對數(shù)據(jù)進行預處理后發(fā)送給Kafka，接下來Spark Streaming再接收gender數(shù)據(jù)進行后續(xù)處理。

3.3 實時數(shù)據(jù)模擬模塊設計

該模塊需要啟動Kafka服務，實例化多個Kafka生產(chǎn)者，用于讀取用戶日志文件。每次讀取一行或多行，對關注的數(shù)據(jù)進行清洗后，每隔固定時間發(fā)送給Spark Streaming實時處理模塊，這樣在固定時間間隔內(nèi)，可發(fā)送一定數(shù)量的購物日志，以便后期進行可視化展示。

3.4 Spark Streaming實時處理模塊設計

該模塊為整個系統(tǒng)的核心模塊，基于Spark流計算的模式按秒來處理Kafka發(fā)送來的數(shù)據(jù)流。利用Spark Streaming接口reduceByKeyAndWindow，設置窗口大小和滑動步長進行數(shù)據(jù)處理、格式轉(zhuǎn)換等操作，并把檢查點文件寫入分布式文件系統(tǒng)HDFS。最后，再實例化一個KafkaProducer實例，用于向Kafka投遞消息，這里發(fā)送數(shù)據(jù)的topic為result。由于Web端最后可視化展示所需要的數(shù)據(jù)又來自Kafka，所以可以實例化一個KafkaConsumer來測試實時處理模塊是否已經(jīng)把處理后的數(shù)據(jù)發(fā)送到“result”的topic上，測試結(jié)果如圖4所示。

圖4 測試結(jié)果

3.5 可視化展示模塊設計

本模塊主要利用Flask-SocketIO接收來自kafka的處理后的數(shù)據(jù)，并將結(jié)果實時推送到瀏覽器端，同時存入MySQL數(shù)據(jù)庫進行數(shù)據(jù)備份。通過編寫對應的html文件以及js文件，實時接收服務端的消息，利用pycharts在web端繪制可交互的圖表，將結(jié)果實時進行展示。值得一提的是，該可視化圖表均為動態(tài)展示效果，隨著時間變化實時數(shù)據(jù)改變，圖表也隨之而變化。后期，針對不同的數(shù)據(jù)，均可進行不同形式的可視化展示。部分效果圖如圖5、圖6所示。

圖5 效果圖1

圖6 效果圖2

4 結(jié)語

綜上所述，本文基于分布式大數(shù)據(jù)技術，設計并實現(xiàn)了電商信息實時展示系統(tǒng)，可以實時模擬流數(shù)據(jù)，并借助Spark Streaming對數(shù)據(jù)進行處理，并做可視化動態(tài)展示。針對系統(tǒng)的優(yōu)化問題，參考了Spark框架中的部分開源代碼，進行了分析測試，選用了高性能算子，替代常規(guī)的算子，達到性能上的提升。通過Spark Streaming、Kafka、WebSocket等技術，可將此開發(fā)模式復制到其他類似應用場景，對海量數(shù)據(jù)實時處理提供了一種可行的解決方案。