摘 要：工業(yè)大數(shù)據(jù)具有規(guī)模龐大、業(yè)務(wù)復(fù)雜等的特點，為數(shù)據(jù)存儲、查詢和分析計算帶了難度。為了優(yōu)化工業(yè)大數(shù)據(jù)存儲管理，提高系統(tǒng)存儲、查詢、分析效率，利用基于Hadoop技術(shù)針對業(yè)務(wù)庫和實時監(jiān)控數(shù)據(jù)庫的存儲管理進行優(yōu)化。系統(tǒng)設(shè)計業(yè)務(wù)庫的集群化同步存儲架構(gòu)，基于Maxwell組件將MySQL業(yè)務(wù)庫數(shù)據(jù)實時同步到HBase，實現(xiàn)業(yè)務(wù)庫的讀寫分離、提高數(shù)據(jù)查詢和數(shù)據(jù)分析的效率;其次，基于Kafka和Flink對業(yè)務(wù)庫同步數(shù)據(jù)進行實時計算處理，實現(xiàn)高并發(fā)數(shù)據(jù)寫入場景下的低延遲響應(yīng);最后，實驗進行了HBase和MySQL的性能對比測試，結(jié)果表明本系統(tǒng)在大規(guī)模數(shù)據(jù)場景下具有更好的計算效率表現(xiàn)，能夠有效進行工業(yè)大數(shù)據(jù)分析存儲。

　　關(guān)鍵詞：工業(yè)大數(shù)據(jù);Hadoop;Flink;HBase

　　引言

　　工業(yè)數(shù)據(jù)的存儲分析是工業(yè)信息化應(yīng)用、推進智能制造的前提和基礎(chǔ)[1]，然而工業(yè)數(shù)據(jù)的海量性、增量性為其的存儲管理帶來了難度，同時也對數(shù)據(jù)存儲的可拓展性、高效性提出了高要求[2]。目前，大多工業(yè)信息系統(tǒng)[3-4]通過結(jié)構(gòu)化數(shù)據(jù)庫如MySQL等進行數(shù)據(jù)存儲。面對頻繁讀寫的應(yīng)用服務(wù)，有研究[4]通過備份同步業(yè)務(wù)庫，實現(xiàn)讀寫分離的架構(gòu)，從而減輕數(shù)據(jù)庫壓力。然而，這種存儲管理方式對于復(fù)雜業(yè)務(wù)表的數(shù)據(jù)分析方面并不友好，需要通過垂直切分或者水平切分進行數(shù)據(jù)查詢。

　　大數(shù)據(jù)存儲系統(tǒng)HBase是一種分布式的列式數(shù)據(jù)庫，針對復(fù)雜業(yè)務(wù)的分析具有天然的優(yōu)勢，被廣泛地應(yīng)用在數(shù)據(jù)存儲和分析過程中[5-8]。然而，HBase的存儲應(yīng)用很難直接切入到現(xiàn)有系統(tǒng)中，或是需要將整套技術(shù)方案推翻重來。同時，不同于普通應(yīng)用系統(tǒng)，工業(yè)數(shù)據(jù)因其特殊的應(yīng)用場景會產(chǎn)生大量的實時監(jiān)控數(shù)據(jù)[2]，如設(shè)備、儀表、定位等。這些實時增量不斷增長的時序數(shù)據(jù)為數(shù)據(jù)存儲的效率提出了要求。此外，在數(shù)萬臺機器毫秒級監(jiān)控的場景中，服務(wù)器每秒需要處理GB級的數(shù)據(jù)，傳統(tǒng)通過負載均衡進行實時計算的處理方式已經(jīng)達到瓶頸。

　　為此，本文提出了一種工業(yè)大數(shù)據(jù)存儲管理與分析系統(tǒng)，基于Hadoop平臺構(gòu)建數(shù)據(jù)存儲平臺，通過Maxwell實時讀取MySQL的數(shù)據(jù)日志寫入Kafka消息隊列，并通過Flink消費處理同步到HBase，在不影響當前系統(tǒng)業(yè)務(wù)庫的同時提高數(shù)據(jù)查詢和存儲管理效率。

　　1 相關(guān)工作

　　1.1 Hadoop平臺簡介

　　從狹義上來說，Hadoop[5-8]是一個由Apache基金會所維護的分布式系統(tǒng)基礎(chǔ)架構(gòu)，而從廣義上來說，Hadoop通常指的是它所構(gòu)建的Hadoop生態(tài)，包括Hadoop核心技術(shù)以及基于Hadoop平臺所部署的大數(shù)據(jù)開源組件和產(chǎn)品。這些組件實現(xiàn)大數(shù)據(jù)場景下的數(shù)據(jù)存儲、分布式計算、數(shù)據(jù)分析、實時計算、數(shù)據(jù)傳輸?shù)取?/p>

　　Hadoop的核心技術(shù)：HDFS、MapReduce、HBase被譽為Hadoop的三駕馬車，更為企業(yè)生產(chǎn)應(yīng)用帶來了高可靠、高容錯和高效率等特性。其中，HBase是一個可伸縮、分布式、面向列的數(shù)據(jù)庫，和傳統(tǒng)關(guān)系數(shù)據(jù)庫不同，HBase提供了對大規(guī)模數(shù)據(jù)的隨機、實時讀寫訪問，同時，HBase中保存的數(shù)據(jù)可以使用MapReduce來處理，它將數(shù)據(jù)存儲和并行計算完美地結(jié)合在一起。

　　1.2 Flink引擎簡介

　　Flink[9]是一個基于內(nèi)存計算的分布式計算框架，通過基于流式計算模型對有界和無界數(shù)據(jù)提供批處理和流處理計算。在實時計算方面，相比于開源方案Storm和Spark Streaming，F(xiàn)link能夠提供準實時的數(shù)據(jù)計算，并能夠?qū)⑴幚砗土魈幚斫y(tǒng)一，實現(xiàn)“批流一體”的整體化方案。這種架構(gòu)使得Flink在執(zhí)行計算時具有較低的延遲，F(xiàn)link被譽為繼Hadoop、Spark之后的第三代分布式計算引擎。

　　1.3 Maxwell簡介

　　Maxwell是一個能實時讀取MySQL二進制日志binlog、并生成json格式的消息，作為生產(chǎn)者發(fā)送給Kafka、RabbitMQ、Redis、文件或其它平臺的應(yīng)用程序。目前，常用的binlog解析工具還有canal、MySQL_streamer，canal由Java開發(fā)，性能穩(wěn)定，但需要自己編寫客戶端來消費canal解析到的數(shù)據(jù);MySQL_streamer由Python開發(fā)，但其技術(shù)文檔比較粗略，對開發(fā)過程并不友好。

　　2 系統(tǒng)總體設(shè)計

　　系統(tǒng)架構(gòu)設(shè)計：為了實現(xiàn)大規(guī)模工業(yè)數(shù)據(jù)的高效存儲，設(shè)計基于Hadoop的工業(yè)大數(shù)據(jù)存儲管理系統(tǒng)總體架構(gòu)，共包括前端集群、后端業(yè)務(wù)集群和數(shù)據(jù)計算集群，具體存儲系統(tǒng)架構(gòu)如圖2所示。

　　系統(tǒng)主要采用前端界面和后端業(yè)務(wù)分離的思想，在前端集群中，由Nginx負責請求的反向代理和負載均衡，分別指向靜態(tài)文件服務(wù)器或Web服務(wù)器，實現(xiàn)網(wǎng)頁相關(guān)界面的顯示與交互。前端集群通過遠程調(diào)用的方式與后端業(yè)務(wù)集群進行通信，實現(xiàn)相關(guān)業(yè)務(wù)操作、MySQL數(shù)據(jù)庫交互操作、數(shù)據(jù)計算與結(jié)果緩存到Redis等操作。對于后端業(yè)務(wù)操作中的數(shù)據(jù)計算環(huán)節(jié)則由數(shù)據(jù)計算集群負責，如：實時同步業(yè)務(wù)庫、設(shè)備數(shù)據(jù)實時計算等。

　　在數(shù)據(jù)計算集群中部署了Hadoop平臺(HDFS、HBase、Yarn)以及Flink、Kafka、Zookeeper等組件。其中HDFS負責進行底層數(shù)據(jù)的存儲，具體由HDFS的DataNode進行文件分片多備份存放，由NameNode進行元數(shù)據(jù)管理和文件操作管理，同時通過Zookeeper注冊兩個NameNode并實時監(jiān)控狀態(tài)，防止一方故障立即切換到另一個，從而保證NameNode的高可用性。HBase負責對同步業(yè)務(wù)庫和時序數(shù)據(jù)庫進行存儲，由HMaster管理多個RegionServer進行數(shù)據(jù)維護和查詢，底層由HDFS進行存儲。對于實時計算部分通過Kafka Broker接受Kafka生產(chǎn)者生產(chǎn)的實時消息，再通過Kafka消費者Flink進行處理計算，其中Kafka的生產(chǎn)、消費進度由Zookeeper進行記錄。Flink不僅提供實時計算，同時提供離線批量計算，其計算過程通過Yarn申請計算資源，具體由ResourceManager管理資源并分配到NodeManager上進行計算。

　　3 工業(yè)大數(shù)據(jù)存儲管理系統(tǒng)

　　3.1 基于Maxwell的業(yè)務(wù)庫同步設(shè)計

　　為了緩解基礎(chǔ)業(yè)務(wù)庫的讀寫壓力，提高復(fù)雜業(yè)務(wù)表的查詢分析效率，系統(tǒng)利用Maxwell實時監(jiān)聽MySQL的binlog日志，然后解析成json格式發(fā)到消息隊列Kafka，再通過Flink消費Kafka數(shù)據(jù)存儲到HBase，從而供其他后端分析業(yè)務(wù)進行讀取、查詢。基于Maxwell的業(yè)務(wù)庫同步設(shè)計具體過程如圖3所示。

　　其具體實現(xiàn)步驟如下：

　　(1)編輯MySQL配置文件my.cnf，開啟binlog功能;

　　(2)創(chuàng)建Maxwell用戶并賦權(quán)限;

　　(3)啟動Kafka集群;

　　(4)修改Maxwell的config.properties文件，配置MySQL數(shù)據(jù)庫連接信息、配置producer類型為Kafka、配置Kafka集群連接信息和topic、配置同步業(yè)務(wù)庫信息;

　　(5)啟動Maxwell，開始監(jiān)聽;

　　(6)創(chuàng)建Flink消費Kafka任務(wù)，對Maxwell產(chǎn)生的數(shù)據(jù)進行實時處理寫入HBase。

　　3.2 基于Kafka和Flink的實時計算

　　對于實時同步的MySQL業(yè)務(wù)庫binlog數(shù)據(jù)，Maxwell首先進行解析傳入Kafka消息隊列，然后通過Flink對這些實時產(chǎn)生的業(yè)務(wù)庫同步數(shù)據(jù)進行消費，實現(xiàn)寫入HBase中。具體步驟包括：

　　(1)在Kafka中創(chuàng)建消息訂閱主題“maxwell”，定義副本數(shù)2個，分區(qū)數(shù)9個。Maxwell作為生產(chǎn)者對MySQL的binlog文件進行解析成json格式數(shù)據(jù)，再發(fā)送到“maxwell”這個主題下。

　　(2)服務(wù)器端配置連接信息，包括：Flink流式處理環(huán)境、Zookeeper的集群信息、Kafka集群信息、消費者組信息、數(shù)據(jù)格式等。

　　(3)通過Kafka Flink Connector API創(chuàng)建線程池對接Kafka，將Maxwell的同步數(shù)據(jù)實時寫入HBase。通過Flink的DataStream算子的map過程處理每一條消息，分別調(diào)用HBase API執(zhí)行數(shù)據(jù)寫入操作。

　　4 系統(tǒng)實現(xiàn)

　　4.1 集群環(huán)境部署

　　系統(tǒng)在1個主節(jié)點、6個計算節(jié)點上搭建Hadoop集群，同時部署MySQL主備節(jié)點、Kafka、Flink、Maxwell等組件。各節(jié)點配置包括：CentOS 7.3 64位操作系統(tǒng)、Intel(R) Xeon CPU 2.4GHz 4Core的CPU、24GB內(nèi)存、1TB硬盤，Hadoop版本為Hadoop 2.6.0，F(xiàn)link版本為Flink 1.9.0，MySQL版本為MySQL 5.6。

　　推薦閱讀：工業(yè)控制系統(tǒng)論文怎么發(fā)表SCI期刊