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

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

　　引言

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

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

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

　　1 相關(guān)工作

　　1.1 Hadoop平臺(tái)簡(jiǎn)介

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

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

　　1.2 Flink引擎簡(jiǎn)介

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

　　1.3 Maxwell簡(jiǎn)介

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

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

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

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

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

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

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

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

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

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

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

　　(3)啟動(dòng)Kafka集群;

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

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

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

　　3.2 基于Kafka和Flink的實(shí)時(shí)計(jì)算

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

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

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

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

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

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

　　系統(tǒng)在1個(gè)主節(jié)點(diǎn)、6個(gè)計(jì)算節(jié)點(diǎn)上搭建Hadoop集群，同時(shí)部署MySQL主備節(jié)點(diǎn)、Kafka、Flink、Maxwell等組件。各節(jié)點(diǎn)配置包括：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期刊