問(wèn)題描述

電子商務(wù)門(mén)戶(hù)(http://www.aaaa.com)希望構(gòu)建一個(gè)實(shí)時(shí)分析儀表盤(pán)，對(duì)每分鐘發(fā)貨的訂單數(shù)量做到可視化，從而優(yōu)化物流的效率。

解決方案

解決方案之前，先快速看看我們將使用的工具：

Apache Spark – 一個(gè)通用的大規(guī)模數(shù)據(jù)快速處理引擎。Spark的批處理速度比Hadoop MapReduce快近10倍，而內(nèi)存中的數(shù)據(jù)分析速度則快近100倍。

Python – Python是一種廣泛使用的高級(jí)，通用，解釋?zhuān)瑒?dòng)態(tài)編程語(yǔ)言。

Kafka – 一個(gè)高吞吐量，分布式消息發(fā)布訂閱系統(tǒng)。

Node.js – 基于事件驅(qū)動(dòng)的I/O服務(wù)器端JavaScript環(huán)境，運(yùn)行在V8引擎上。

Socket.io – Socket.IO是一個(gè)構(gòu)建實(shí)時(shí)Web應(yīng)用程序的JavaScript庫(kù)。它支持Web客戶(hù)端和服務(wù)器之間的實(shí)時(shí)、雙向通信。

Highcharts – 網(wǎng)頁(yè)上交互式JavaScript圖表。

CloudxLab – 提供一個(gè)真實(shí)的基于云的環(huán)境，用于練習(xí)和學(xué)習(xí)各種工具。

如何構(gòu)建數(shù)據(jù)Pipeline?

下面是數(shù)據(jù)Pipeline高層架構(gòu)圖

數(shù)據(jù)Pipeline

實(shí)時(shí)分析Dashboard

讓我們從數(shù)據(jù)Pipeline中的每個(gè)階段的描述開(kāi)始，并完成解決方案的構(gòu)建。

階段1

當(dāng)客戶(hù)購(gòu)買(mǎi)系統(tǒng)中的物品或訂單管理系統(tǒng)中的訂單狀態(tài)變化時(shí)，相應(yīng)的訂單ID以及訂單狀態(tài)和時(shí)間將被推送到相應(yīng)的Kafka主題中。

數(shù)據(jù)集

由于沒(méi)有真實(shí)的在線(xiàn)電子商務(wù)門(mén)戶(hù)網(wǎng)站，我們準(zhǔn)備用CSV文件的數(shù)據(jù)集來(lái)模擬。讓我們看看數(shù)據(jù)集：

數(shù)據(jù)集包含三列分別是：“DateTime”、“OrderId”和“Status”。數(shù)據(jù)集中的每一行表示特定時(shí)間時(shí)訂單的狀態(tài)。這里我們用“xxxxx-xxx”代表訂單ID。我們只對(duì)每分鐘發(fā)貨的訂單數(shù)感興趣，所以不需要實(shí)際的訂單ID。

可以從CloudxLab GitHub倉(cāng)庫(kù)克隆完整的解決方案的源代碼和數(shù)據(jù)集。

數(shù)據(jù)集位于項(xiàng)目的spark-streaming/data/order_data文件夾中。

推送數(shù)據(jù)集到Kafka

shell腳本將從這些CSV文件中分別獲取每一行并推送到Kafka。推送完一個(gè)CSV文件到Kafka之后，需要等待1分鐘再推送下一個(gè)CSV文件，這樣可以模擬實(shí)時(shí)電子商務(wù)門(mén)戶(hù)環(huán)境，這個(gè)環(huán)境中的訂單狀態(tài)是以不同的時(shí)間間隔更新的。在現(xiàn)實(shí)世界的情況下，當(dāng)訂單狀態(tài)改變時(shí)，相應(yīng)的訂單詳細(xì)信息會(huì)被推送到Kafka。

運(yùn)行我們的shell腳本將數(shù)據(jù)推送到Kafka主題中。登錄到CloudxLab Web控制臺(tái)并運(yùn)行以下命令。

階段2

在第1階段后，Kafka“order-data”主題中的每個(gè)消息都將如下所示

階段3

Spark streaming代碼將在60秒的時(shí)間窗口中從“order-data”的Kafka主題獲取數(shù)據(jù)并處理，這樣就能在該60秒時(shí)間窗口中為每種狀態(tài)的訂單計(jì)數(shù)。處理后，每種狀態(tài)訂單的總計(jì)數(shù)被推送到“order-one-min-data”的Kafka主題中。

請(qǐng)?jiān)赪eb控制臺(tái)中運(yùn)行這些Spark streaming代碼

階段4

在這個(gè)階段，Kafka主題“order-one-min-data”中的每個(gè)消息都將類(lèi)似于以下JSON字符串

階段5

運(yùn)行Node.js server

現(xiàn)在我們將運(yùn)行一個(gè)node.js服務(wù)器來(lái)使用“order-one-min-data”Kafka主題的消息，并將其推送到Web瀏覽器，這樣就可以在Web瀏覽器中顯示出每分鐘發(fā)貨的訂單數(shù)量。

請(qǐng)?jiān)赪eb控制臺(tái)中運(yùn)行以下命令以啟動(dòng)node.js服務(wù)器

現(xiàn)在node服務(wù)器將運(yùn)行在端口3001上。如果在啟動(dòng)node服務(wù)器時(shí)出現(xiàn)“EADDRINUSE”錯(cuò)誤，請(qǐng)編輯index.js文件并將端口依次更改為3002...3003...3004等。請(qǐng)使用3001-3010范圍內(nèi)的任意可用端口來(lái)運(yùn)行node服務(wù)器。

用瀏覽器訪(fǎng)問(wèn)

啟動(dòng)node服務(wù)器后，請(qǐng)轉(zhuǎn)到http://YOUR_WEB_CONSOLE:PORT_NUMBER訪(fǎng)問(wèn)實(shí)時(shí)分析Dashboard。如果您的Web控制臺(tái)是f.cloudxlab.com，并且node服務(wù)器正在端口3002上運(yùn)行，請(qǐng)轉(zhuǎn)到http://f.cloudxlab.com:3002訪(fǎng)問(wèn)Dashboard。

當(dāng)我們?cè)L問(wèn)上面的URL時(shí)，socket.io-client庫(kù)被加載到瀏覽器，它會(huì)開(kāi)啟服務(wù)器和瀏覽器之間的雙向通信信道。

階段6

一旦在Kafka的“order-one-min-data”主題中有新消息到達(dá)，node進(jìn)程就會(huì)消費(fèi)它。消費(fèi)的消息將通過(guò)socket.io發(fā)送給Web瀏覽器。

階段7

一旦web瀏覽器中的socket.io-client接收到一個(gè)新的“message”事件，事件中的數(shù)據(jù)將會(huì)被處理。如果接收的數(shù)據(jù)中的訂單狀態(tài)是“shipped”，它將會(huì)被添加到HighCharts坐標(biāo)系上并顯示在瀏覽器中。

截圖

我們還錄制了一個(gè)關(guān)于如何運(yùn)行上述所有的命令并構(gòu)建實(shí)時(shí)分析Dashboard的視頻。

我們已成功構(gòu)建實(shí)時(shí)分析Dashboard。這是一個(gè)基本示例，演示如何集成Spark-streaming，Kafka，node.js和socket.io來(lái)構(gòu)建實(shí)時(shí)分析Dashboard?，F(xiàn)在，由于有了這些基礎(chǔ)知識(shí)，我們就可以使用上述工具構(gòu)建更復(fù)雜的系統(tǒng)。