基于Storm的工業(yè)流水線實時分析系統(tǒng)設(shè)計與實現(xiàn)

陳志云 肖楚喬

(華東師范大學計算機科學與軟件工程學院 上海 200062)

基于Storm的工業(yè)流水線實時分析系統(tǒng)設(shè)計與實現(xiàn)

陳志云 肖楚喬*

(華東師范大學計算機科學與軟件工程學院 上海 200062)

隨著全球云計算、大數(shù)據(jù)、物聯(lián)網(wǎng)和人工智能等技術(shù)在工業(yè)領(lǐng)域的興起，數(shù)據(jù)成為了工業(yè)4.0時代的核心驅(qū)動力。為了彌補Hadoop分布式系統(tǒng)在實時工業(yè)數(shù)據(jù)處理中顯現(xiàn)的不足，提出基于Storm的工業(yè)流水線實時分析系統(tǒng)。該系統(tǒng)通過嵌入SDK實時采集終端數(shù)據(jù)，利用Nginx服務(wù)器將數(shù)據(jù)轉(zhuǎn)換成日志文件，并采用分布式消息系統(tǒng)Kafka緩存，消息流入Storm進行分析處理后，將結(jié)果存入HBase中。最后從系統(tǒng)的保障性，并行性和實時性等方面進行分析，表明該系統(tǒng)非常可靠地將各個終端收集到的工業(yè)數(shù)據(jù)實時轉(zhuǎn)換成有價值的信息輸出，方便數(shù)據(jù)資源的匯總與優(yōu)化。

工業(yè)大數(shù)據(jù) Storm Kafka

0 引 言

近年來，各發(fā)達國家均從戰(zhàn)略層面提出了一系列大數(shù)據(jù)技術(shù)研發(fā)計劃，我國也在2013年提出了《大數(shù)據(jù)技術(shù)與產(chǎn)業(yè)發(fā)展白皮書》[1-2]。并且隨著德國“工業(yè)4.0”，美國“制造業(yè)回歸”和“中國制造2025”的提出，各行各業(yè)的決策正在從“業(yè)務(wù)驅(qū)動"轉(zhuǎn)變到"數(shù)據(jù)驅(qū)動”[3-4]。如何利用大數(shù)據(jù)分析技術(shù)將得到的工業(yè)數(shù)據(jù)再創(chuàng)造成新的知識與價值，成為了第四次工業(yè)革命的最終目標之一[5-6]。面對設(shè)備控制器、制造系統(tǒng)、流水線等設(shè)備層產(chǎn)生的更為復雜多源、實時易變的海量工業(yè)數(shù)據(jù)，如何實時地將各個地區(qū)的工業(yè)數(shù)據(jù)上傳到上層的制造執(zhí)行管理系統(tǒng)MES，幫助企業(yè)優(yōu)化資源配置，整合分散在各個系統(tǒng)中的數(shù)據(jù)資源，提供智能預測，已成為制造業(yè)面臨的新挑戰(zhàn)[7-8]。

工業(yè)大數(shù)據(jù)可廣泛應用于企業(yè)的整個生產(chǎn)過程，例如研發(fā)設(shè)計、供應鏈、生產(chǎn)制造、營銷與服務(wù)環(huán)節(jié)。在設(shè)備仿真階段，瑪莎拉蒂通過數(shù)字化工具加速產(chǎn)品設(shè)計，將研發(fā)效率提升30%。在生產(chǎn)制造環(huán)節(jié)中，通過生產(chǎn)線、生產(chǎn)設(shè)備等抓取數(shù)據(jù)，再利用無線通信傳輸數(shù)據(jù)，從而對生產(chǎn)本身實時監(jiān)控[9-11]。不僅如此，IBM、EMC、華為等企業(yè)也都在開源大數(shù)據(jù)軟件之上研發(fā)面向制造業(yè)的轉(zhuǎn)型升級的大數(shù)據(jù)產(chǎn)品和系統(tǒng)[12]。隨著國內(nèi)外不斷出現(xiàn)許多基于Storm的實時分布式流處理計算平臺，例如：360實時平臺、新浪實時分析平臺、騰訊實時計算平臺、阿里JStorm等[13-14]，加之Storm還可以很方便地集成現(xiàn)有的各種技術(shù)框架，例如Kafka、HBase、HDFS等，使得Storm的應用領(lǐng)域和場景越來越廣泛。

1 實時流處理相關(guān)技術(shù)

1.1 流式計算框架Storm

由Twitter公司開源的Storm，被廣泛應用于實時分析統(tǒng)計，在線機器學習、持續(xù)計算、ETL等領(lǐng)域。Storm的出現(xiàn)解決了Hadoop不能處理實時數(shù)據(jù)的問題。但是Storm與Hadoop同樣都是主從系統(tǒng)架構(gòu)，由一個主節(jié)點運行Nimbus進程和多個從節(jié)點運行Supervisor進程，并且使用Zookeeper協(xié)作框架協(xié)調(diào)工作[15-16]。

1.1.1 Storm數(shù)據(jù)流模型Topology

在Storm中數(shù)據(jù)流stream可以抽象視為不間斷且連續(xù)的tuple元組。由spout組件作為tuples的發(fā)送者，根據(jù)數(shù)據(jù)源的不同，spout可以訂制化開發(fā)。bolt組件稱為處理器，每個bolt可以消費任意數(shù)量的tuples，也可以發(fā)送給別的bolt。為了提高處理的效率，一般會有多個spout和bolt。Storm將此數(shù)據(jù)流模型稱為Topology，如圖1所示。Topology和Hadoop的job不同之處在于：job執(zhí)行完后會主動停止，而Topology不會自動停止。

圖1 拓撲模型示例

1.1.2 Storm框架的主從架構(gòu)

Nimbus作為主節(jié)點的守護進程，監(jiān)聽Client提交的Topology。同時根據(jù)worker資源情況，計算每一個task應該如何在worker中分配，并且在Zookeeper上存儲task和worker的對應信息。Supervisor作為Storm集群從節(jié)點上的運行進程，定時從Zookeeper上檢查并獲取最新的tasks到worker中。每個Supervisor可以有多個worker進程，worker進程又會啟動多個executor去執(zhí)行task。實際中executor的并發(fā)數(shù)量就可以代表Storm集群的并行度。Zookeeper在完成主從節(jié)點協(xié)調(diào)工作的同時，還創(chuàng)建節(jié)點監(jiān)控task的心跳，如圖2所示。

圖2 Storm主從架構(gòu)圖

1.2 分布式消息系統(tǒng)Kafka

Kafka是一種分布式消息發(fā)布與訂閱架構(gòu)，常用于網(wǎng)頁行為跟蹤、運營數(shù)據(jù)監(jiān)控、日志聚合和流處理等場景[17]。Kafka中的主題Topic是可以被多個消費者共享的隊列，并且可以有多個partition，每個partition分布在不同的代理Broker中。Producer作為消息的生產(chǎn)者，根據(jù)指定的partition和消息id，將消息push到指定的partition中。Consumer作為消息的消費者，通過pull的方式從Broker中訂閱并消費消息，流程如圖3所示。

圖3 Kafka消息流程圖

1.3 文件收集庫框架Flume

Flume是僅基于Linux環(huán)境的分布式日志數(shù)據(jù)收集服務(wù)框架[18]。它是一個簡單的可擴展的數(shù)據(jù)模型，常用于實時數(shù)據(jù)分析。Flume中events是數(shù)據(jù)傳輸?shù)幕締卧?，Source通過監(jiān)控某個文件，拿到日志數(shù)據(jù)封裝到event中，并且put到channel隊列中。Sink會主動到channel中拉取數(shù)據(jù)，再寫到下一個地方，如圖4所示。

圖4 Flume框架流程圖

2 基于Storm的工廠流水線實時分析系統(tǒng)

目前工業(yè)生產(chǎn)過程正在不斷地邁向智能化和信息化時代，如何對工業(yè)流水線的高效管理，將成為企業(yè)市場競爭力的保障[19]。但是目前的流水線管理方面仍然存在信息容量有限，無法及時更新信息，易丟失等缺點[20]。針對流水線具有實時和大量的特點，提出一種基于Storm的工廠流水線實時分析系統(tǒng)：從移動端或PC端實時上傳流水線數(shù)據(jù)，生成Nginx日志，通過Storm的Topology實時分析各個廠區(qū)和各類型生產(chǎn)線上的生產(chǎn)總量、產(chǎn)品類型、終端編號等信息。并且設(shè)計了基于Storm的ack保障機制可以防止數(shù)據(jù)的丟失，使HBase可以實現(xiàn)持久化的存儲。操作人員只需要在終端輸入一些簡單的流水線信息，企業(yè)管理人員便可以直接看到各地區(qū)工廠流水線生產(chǎn)狀況。

2.1 系統(tǒng)框架

本文設(shè)計的系統(tǒng)采用Storm+Kafka+Flume+Nginx+HBase的實時流處理框架，流水線數(shù)據(jù)會在終端被輸入的同時轉(zhuǎn)換成日志格式。日志通過Flume讀取到Kafka的topic主題中，并且此主題存在Zookeeper上。然后設(shè)計Kafka-Spout不斷向Storm發(fā)送日志記錄，通過Topology對記錄分析，并將分析結(jié)果持久化保存到HBase中。

(1) 數(shù)據(jù)收集層：本層屬于數(shù)據(jù)源層，通過不同的嵌入SDK創(chuàng)建event采集需要的流水線的數(shù)據(jù)信息。例如：流水線類型和編號、產(chǎn)品類型和數(shù)量、廠區(qū)編號、出現(xiàn)的問題等數(shù)據(jù)。

(2) 日志生成層：通過Nginx服務(wù)器，將獲取到的數(shù)據(jù)轉(zhuǎn)成日志文件，并且按照時間點進行日志分割。日志格式定義：

log_format access ′remote_addr remote_user [time_local] ″request″ status body_bytes_sent ″http_referer″ ″http_user_agent″ ′;

(3) 生產(chǎn)者層：通過Flume Agent監(jiān)控Nginx的日志文件，通過tail命令不斷讀取日志文件，并傳輸?shù)終afka的指定Topic中。

(4) 主題隊列層：本層是Kafka集群，因為具備水平擴展和高吞吐量等特性，用來保障輸入數(shù)據(jù)的負載均衡和消息緩沖。該層的設(shè)計是防止數(shù)據(jù)采集端和Storm處理速度不同而導致的數(shù)據(jù)丟失。

(5) 業(yè)務(wù)邏輯層：構(gòu)建Topology，對日志文件中每條記錄進行拆分、分流、匯總，并實時發(fā)送到HBase中持久化存儲。

(6) 持久化存儲：選擇HBase，因為其適用于海量數(shù)據(jù)存儲和實時查詢，并且有良好的兼容性，支持多種大數(shù)據(jù)框架。

流水線實時分析框架如圖5所示。

圖5 基于Storm的工廠流水線實時分析框架

2.2 Storm業(yè)務(wù)邏輯層的設(shè)計

2.2.1 Topology方案設(shè)計

為了增加消息發(fā)送的靈活度，Storm中通過Streamgrouping定義一個stream應該如何分配bolt上的多個task。在Storm中有6種類型的流發(fā)送方式：Shuffle grouping、Fields grouping、All grouping、Global grouping、Non grouping和Direct grouping。Shuffle表示隨機分組，可保證每個bolt接收到的tuple數(shù)基本相等。Fields表示按字段分組，具有相同id的tuple會被發(fā)送到相同的bolt中。Global表示全局分組，Tuple只會被分配到一個bolt中。

針對Storm的拓撲模型和流分發(fā)策略，設(shè)計了如圖6所示的拓撲結(jié)構(gòu)。

圖6 流水線實時分析Topology

此拓撲共分為5個階段。

(1) 由于所有的日志記錄都存在Kafka集群的消息隊列中，因此設(shè)計Spout采用KafkaSpout。直接引用storm.kafka.*包，創(chuàng)建實例化KafkaSpout，通過提供的topic參數(shù)讀取zookeeper上的主題位置。采用shufflegrouping的方式將記錄隨機發(fā)送給LogSplitBolt。

(2) LogSplitBolt對發(fā)送來的tuple進行解析，抽取各個字段。然后按照Stream設(shè)定的id號采用fieldsgroup流發(fā)送方式給HttpSplitBolt、IPCountBolt、AgentSplitBolt。

(3) HttpSplitBolt是將URL中傳入的流水線參數(shù)和產(chǎn)品參數(shù)抽取出來，按照Stream設(shè)定的id號fieldsgroup發(fā)送給LineCountBolt和ProdCountBolt。同理，AgentSplitBolt將useragent中傳入的browser和os信息抽取出來分id發(fā)送給BrowCountBolt和OSCountBolt。

(4) IPCountBolt、LineCountBolt、ProdCountBolt、BrowCountBolt、OSCountBolt作為計數(shù)模塊，統(tǒng)計各個id的次數(shù)。并采用globalgrouping流發(fā)送方式到HBaseBolt。

(5) HBaseBolt設(shè)計。首先在prepare函數(shù)中創(chuàng)建Configuration對象并加載配置文件hbase-site.xml，連接數(shù)據(jù)庫；然后再在execute函數(shù)中設(shè)計相應字段名稱put到HBase中。

2.2.2 Tuple樹的設(shè)計

Tuple元組是對Stream的抽象，每一個tuple可以包含多列，每列都是一個，且Topology在設(shè)計的時候必須定義tuple的字段類型。類型可以是：Integer、long、short、byte、String、double、float、byte、array和boolean。針對各組件輸入流和輸出流的類型設(shè)計出如圖7所示Tuple樹。

圖7 Tuple樹

此Tuple樹是數(shù)據(jù)流Stream在Topology中的抽象表示。首先KafkaSpout將每一條日志記錄封裝成Tuple a格式流入到LogSplitBolt中。LogSplitBolt中設(shè)計正則表達式來抽取需要的字段，再構(gòu)建并流出 Tuple b1、b2、b3。然后由HttpSplitBolt將Tuple b2重新構(gòu)建成Tuple c1 和c2流出。由AgentSplitBolt將Tuple b3重新構(gòu)建成Tuple d1和d2流出。IPCountBolt、LineCountBolt、ProdCountBolt、BrowCountBolt、OSCountBolt將流入的Tuple格式統(tǒng)一重新構(gòu)建成Tuple e，最后流入HBaseBolt中。為了保障每個Tuple都被正確的處理，Storm有ack保障機制，采用異或算法。可以保障這個Tuple以及這個Tuple所產(chǎn)生的子Tuple都被成功的處理。

2.3 HBase設(shè)計

為了將實時統(tǒng)計的數(shù)據(jù)存到HBase中，設(shè)計表Counts。創(chuàng)建：hbase> create ‘Counts’,{Name=>’infos’,VERSION=>2147483647,BLOOMFILTER=>’ROWCOL’}。

將VERSION設(shè)置為Integer類型最大值，可以查看歷史數(shù)據(jù)，如表1所示。

表1 Counts表

3 實驗與分析

實驗環(huán)境：由2臺PC機組成集群；操作系統(tǒng)：CentOS 6.4，內(nèi)存32 GB，JDK1.7，Core i7-6820HQ。集群信息如表2所示。

表2 集群信息

3.1 系統(tǒng)可靠性測試

實驗采用運行Jar包模擬每1 ms產(chǎn)生1條Nginx日志記錄。當數(shù)據(jù)源快速流出tuple時，計算tuple在無保障機制時的丟失率和采用ack保障機制后的tuple丟失率，為了保證準確性，采用3次數(shù)據(jù)的平均值?？梢郧逦乜闯?，保障機制有效地降低了tuple的丟失率，如圖8所示。

圖8 Tuple失敗率

通過上述圖表可以看出該系統(tǒng)能有效地保障流水線數(shù)據(jù)信息的準確性，非常適用于工廠流水線實時生產(chǎn)的模式。

3.2 系統(tǒng)并行度測試

通過Storm UI可以方便地看到Topology的詳細信息。Capacity參數(shù)代表的是拓撲并行度，參數(shù)值越接近1，表示該組件的并行度不夠，需要適當?shù)財U展組件的executor數(shù)量。通常情況下，一個worker里executor數(shù)量必須小于等于給定的task數(shù)量。所以通過改變線程數(shù)量，分析該系統(tǒng)的橫向擴展能力，結(jié)果如圖9所示。

圖9 并行度測試

通過圖9可以看出該系統(tǒng)通過增加線程數(shù)量，確實可以有效降低Capacity數(shù)值，提高系統(tǒng)的并行度。

3.3 系統(tǒng)實時計算性能

通過表3可以得出系統(tǒng)實時計算總量和時間的關(guān)系，從而證明該框架有良好的健壯性和實時性。

表3 處理數(shù)據(jù)量和時間關(guān)系

4 結(jié) 語

隨著國家工業(yè)轉(zhuǎn)型升級的開始，工業(yè)大數(shù)據(jù)的應用將越來越多，而工廠中實時產(chǎn)生的海量數(shù)據(jù)將是創(chuàng)造智慧工廠的關(guān)鍵。本文通過對Storm架構(gòu)的簡單分析，結(jié)合多種開源框架，實現(xiàn)了一個基于Storm的工業(yè)流水線實時分析系統(tǒng)。該系統(tǒng)有良好的保障性和并行性。由于Storm是分布式流計算框架，因此該系統(tǒng)有良好的可擴展性、容錯性、實時性。并且該系統(tǒng)采用開源框架，因此簡單易搭建，且采用Linux環(huán)境具有平臺通用性。接下來的工作是結(jié)合機器學習算法，訓練數(shù)據(jù)實現(xiàn)預測。

[1] 張禮立.數(shù)據(jù)是工業(yè)4.0的核心驅(qū)動[J].中國工業(yè)評論,2015(12):36-43.

[2] 李國杰,程學旗.大數(shù)據(jù)研究:未來科技及經(jīng)濟社會發(fā)展的重大戰(zhàn)略領(lǐng)域——大數(shù)據(jù)的研究現(xiàn)狀與科學思考[J].中國科學院院刊,2012,27(6):5-15.

[3] 羅文.德國工業(yè)4.0戰(zhàn)略對我國工業(yè)轉(zhuǎn)型的啟示[J].玻璃鋼/復合材料,2014(11):125-128.

[4] 李杰,劉宗長.中國制造2025的核心競爭力——挖掘使用數(shù)據(jù)[J].博鰲觀察,2015(4):52-55.

[5] Addo-Tenkorang R,Helo P T.Big Data Applications in Operations/Supply-Chain Management:A Literature Review[J].Computers & Industrial Engineering,2016,101:528-543.

[6] Rousseaux F.BIG DATA and Data-Driven Intelligent Predictive Algorithms to support creativity in Industrial Engineering[J].Computers & Industrial Engineering,2016.

[7] 高嬰勱.工業(yè)大數(shù)據(jù)價值挖掘路徑[J].中國工業(yè)評論,2015(s1):21-27.

[8] 孔憲光,章雄,馬洪波,等.面向復雜工業(yè)大數(shù)據(jù)的實時特征提取方法[J].西安電子科技大學學報自然科學版,2016,43(5):70-74.

[9] 王建民.工業(yè)大數(shù)據(jù)技術(shù)[J].電信網(wǎng)技術(shù),2016(8):1-5.

[10] 單莘,祝智崗,張龍,等.基于流處理技術(shù)的云計算平臺監(jiān)控方案的設(shè)計與實現(xiàn)[J].計算機應用與軟件,2016,33(4):88-90.

[11] 萬英杰,鮑遠松,黃明.分布式工業(yè)數(shù)據(jù)實時分析計算平臺[J].信息技術(shù)與標準化,2016(11):61-63.

[12] 黃明峰.工業(yè)大數(shù)據(jù)發(fā)展態(tài)勢與典型應用[J].電信科學,2016,32(7):175-178.

[13] Sun D W,Zhang G Y,Zheng W M.Big data stream computing:technologies and instances[J].Journal of Software,2014,25(4):839-862.

[14] 孟小峰,慈祥.大數(shù)據(jù)管理:概念、技術(shù)與挑戰(zhàn)[J].計算機研究與發(fā)展,2013(1):146-169.

[15] Hadoop[EB/OL].[2017-01-19].http://hadoop.apache.org/.

[16] Storm[EB/OL].[2017-01-19].https://storm.apache.org/.

[17] Kafka[EB/OL].[2017-01-19].https://kafka.apache.org/.

[18] Flume[EB/OL].[2017-01-19].https://flume.apache.org/.

[19] Zhang Y,Xu J,Sun S,et al.Real-time information driven intelligent navigation method of assembly station in unpaced lines[J].Computers & Industrial Engineering,2015,84(C):91-100.

[20] 李二霞.大數(shù)據(jù)在立柱維修流水線的應用[J].科技創(chuàng)新與應用,2015(30):18-19.

DESIGNOFINDUSTRIALASSEMBLYLINEREAL-TIMEANALYSISSYSTEMBASEDONSTORMANDITSIMPLEMENTATION

Chen Zhiyun Xiao Chuqiao*

(SchoolofComputerScienceandSoftwareEngineering,EastChinaNormalUniversity,Shanghai200062,China)

With the rise of cloud computing, big data, IoT, AI, etc. in the global industrial field, the data has become the core driving force of Industry 4.0 era. In order to make up for the shortage of Hadoop distributed system in real-time industrial data processing, this paper presents an industrial assembly line real-time analysis system based on Storm. The system collects real-time terminal data through the embedded SDK, converts the data to the log file through the Nginx server, and uses the distributed information system Kafka as cache, then the message flowing into Storm and the results stored in HBase. Finally, the analysis of the system from the aspects of security, parallelism and real-time performance shows that the system converts the industrial data collected by each terminal into valuable information to facilitate the aggregation and optimization of data resources reliably.

Industrial big data Storm Kafka

2017-02-05。陳志云，副教授，主研領(lǐng)域：多媒體應用與教育技術(shù)。肖楚喬，碩士生。

TP311

A

10.3969/j.issn.1000-386x.2017.11.009