基于Flink的工業大數據平臺研究與應用

趙潤發，婁淵勝+，葉 楓，石 宏

(1.河海大學 計算機與信息學院，江蘇 南京 211100； 2.南京廣廈軟件有限公司 工業大數據開發部，江蘇 南京 210000)

0 引 言

針對工業大數據平臺的研究[1-4]，文獻[5]采用Dubbo與NoSQL構建了工業領域大數據平臺，為工業領域不斷增長的數據提供了解決辦法。文獻[6]提出了一個面向工業的數據處理系統，其以Spark為框架，選取MySQL和HDFS為存儲介質，實現了工業數據的快速分析。文獻[7]將物聯網與大數據相結合，構建了一個工廠能耗分析平臺，實現了能耗數據的查詢以及數據的分析。

目前工業大數據平臺已得到廣泛研究，但工業大數據平臺技術架構不同，差異性較大，再者對于很多工業大數據平臺而言，其數據處理效率較低，預警時間較長。針對上述問題，本文研究了一種基于Flink的工業大數據平臺，主要貢獻如下：

(1)采用Kafka和Flink進行集成，對數據進行傳輸和處理，并將處理過的數據按照類型存儲至數據池中；

(2)利用Flink對工業大數據進行預處理，提高平臺運行的準確性；

(3)采用多種大數據技術，實現工業大數據平臺的數據查詢以及預警功能，且相對于典型大數據平臺而言，速度更快、效率更高。

1 關鍵技術介紹

1.1 Apache Flink

Apache Flink是一個分布式處理框架，可在無邊界和有邊界數據流上進行計算[8]。Flink不僅能運行在YARN、Mesos等資源管理框架上，而且能在單獨集群中運行，適用于具有不可靠數據源、海量數據處理等場景。此平臺采用Flink的最主要原因是：工業大數據類型雜，既包括流數據，又包括批數據，而Flink兩者都可以處理。它適用的主要場景是流數據方面的，而批數據是“特殊的流數據”，所有任務都可以當成流來處理[8]，并且數據處理延遲性較低。其架構[8]如圖1所示。

對于流數據應用來說，Flink提供DataStream API。對于批數據處理應用來說，提供DataSet API。它支持Java和Scala語言，同時支持Kafka的輸入數據和ElasticSearch、MySQL、InfluxDB多種數據庫。Flink同時具有高度靈活的窗口操作，包括time、count等窗口操作，如：每隔多久發送數據至客戶端、每次發送數據的個數等，十分適用于工業場景。

圖1 Flink架構

1.2 Kafka

Kafka是一個基于Zookeeper的分布式消息系統，它具有高吞吐、低延遲、可靠性好、容錯能力強的良好特性[9]。低延遲體現在Kafka每秒能夠處理巨量信息且延遲很低，只有幾毫秒，適用于工業生產過程中海量數據的處理；高吞吐率體現在即使應用在廉價的商用機器上，Kafka也能進行每秒100 K消息的傳輸。Kafka也較為可靠，傳輸的數據可以在本地磁盤持久保存，同時數據會自動進行備份，數據丟失后仍可找到數據。Kafka容錯性較好，集群中節點是允許失敗的(如副本數量為n,則n-1個節點是允許失敗的)[9]。此平臺選用Kafka消息隊列能夠更好地解耦，也增強了平臺的擴展性，即使企業數據發生改變，不需要改變代碼和調節參數就可以輕松實現用戶要求。同時也保證了數據傳送的順序性和安全性。

1.3 Grafana

Grafana是一個可靠性較好的可視化和度量分析工具。它具有靈活和快捷的客戶端圖表，有多種可視化指標和面板插件，官方庫里有圖表、折線圖、文本文檔等豐富的儀表盤插件；它支持多種數據庫如：MySQL、InfluxDB、Prometheus、OpenTSDB、Elasticsearch和KairosDB等等；Grafana可通過直觀的可視化方式進行預警并發送通知，當獲得的數據大于用戶設定的閾值時通知Slack、DingDing、Email等；并且數據源不同，但仍可以使用在同一圖表中，數據源的來源可以根據每個查詢決定，也可以自定義數據源；Grafana同時具有豐富的注釋圖，注釋圖表能顯示不同數據源的豐富事件，當鼠標停留在圖表時，會以全面的標記來顯示出元數據。

1.4 InfluxDB

InfluxDB是一個用于處理海量數據寫入與數據查詢的時間序列數據庫，應用于有大量時間戳數據的場景下，例如DevOps(過程、方法、系統)監控，物聯網工業數據實時分析等。它是分布式擴展的，不依賴外部任何條件。它還可以對ETL進行后臺處理并實時監控預警。它有類似SQL的查詢語言，可輕松方便查詢到需要的數據。不僅如此，InfluxDB連續查詢自動計算聚合數據，大大提高了頻繁查詢的效率。本平臺中的數據量較大，時間戳數據較多，因此InfluxDB是工業大數據存儲的絕佳選擇。

2 工業大數據平臺框架設計

本平臺旨在實現一個能滿足對工業大數據進行存儲、集成、分析的平臺，能夠為企業多種業務提高指導和決策支持。其架構如圖2所示，其主要分為5個部分，包括：數據源模塊、消息隊列模塊、數據存儲模塊、數據處理模塊、可視化模塊。

2.1 數據源模塊

此平臺的數據源主要分為兩種，一種是靜態系統數據，第二種是實時流數據。數據源獲取的方式主要如下：

(1)靜態系統數據一般是由公司專門人員去收集，如設備生產日期、企業名稱、設備編號等，這些數據以特定的形式整理形成一個Excel表格，能夠直接使用；

(2)大多數的企業獲取數據的方式都是通過各種傳感器，傳感器獲取到的設備的狀態、運行時間等實時數據，然后將這些數據發送給此平臺的處理系統；

(3)企業的很多數據會分布在不同地區的不同公司，所以這時候它們通常會以日志的形式存在，而Flume是一個很好的日志收集工具[10]。這個工具能夠將這些日志文件識別出來，并整理收集在一起，并發往此工業大數據平臺;

(4)工業生產中會產生很多業務靜態數據，但它們的格式可能不是我們所需要的，此時可以使用Sqoop數據源轉換工具，將它們轉換為我們所需要的格式，然后再將這些數據發送給工業大數據平臺。

圖2 體系架構

2.2 消息隊列模塊

消息隊列主要指數據在傳輸過程中保存數據的一個容器。工業大數據類型多，數據量大，面對此場景使用消息隊列是一個極佳的選擇，因為消息隊列能夠極大地降低系統響應時間、提高系統穩定性，同時保證數據傳輸的順序性和安全性，最重要的是實現數據的異步化，并起到解耦的作用。

此模塊選用Kafka作為消息隊列系統，利用Flink將數據源模塊中的實時數據和批數據都暫存至消息隊列中。Flink作為生產者，會源源不斷地生產出消息，然后發送給消息隊列Kafka中，而Kafka就成為了消費者，它會不斷地從Flink中獲取到消息，從而對這些數據進行進一步處理。

2.3 數據處理模塊

本模塊主要采用Flink來處理實時大數據和離線批數據。根據數據類型，將此模塊又分為實時數據處理模塊和批數據處理模塊。Flink能夠同時支持批處理與流處理任務，它包含兩種預先定義的函數：DataStream API和DataSet API。DataStream API 包括reduce、aggregations、filter等方法。DataSet API包括distinct、Hash-Partition、window等方法。

批數據處理模塊中，此平臺會利用aggregations中的sum()、min()、max()方法對批數據進行統計，求出工業數據的最大值、最小值、總和等，并在前端顯示出來。

流數據處理模塊主要是對數據進行預處理。在工業大數據的實際生產過程中，由于人工失誤或者數據采集設備因生產環境惡劣會導致收集到的數據不準確，這些數據如果直接存入到數據庫中不僅會降低大數據平臺查詢數據的準確性，而且會大大降低平臺的運行效率。此模塊主要利用Flink來去除實際業務處理中的無效數據、重復數據以及缺失率較高的數據，其預處理的流程如圖3所示。

圖3 預處理流程

數據預處理方法具體步驟如下：

(1)首先利用Flink從Kafka中獲取到數據，然后通過Flink自帶的RocksDB狀態后端去重方式對工業大數據進行去重。我們需要開啟RocksDB狀態后端并對其參數進行配置，如數據過期的時間、是否過期的數據能再次被訪問等，接著注冊Flink定時器。我們也可以利用Flink的TTL機制，打開RocksDB狀態后端的TTL compaction filter，這樣能在后臺實現重復數據的自動刪除。在處理重復數據時，如果數據的key(如事件ID)對應的狀態不存在，說明此數據沒有出現過，可以更新狀態并且輸出數據。反之，說明此數據已經出現過，RocksDB就會將其自動刪除。同時我們可利用Flink SQL提供的distinct去重方法來統計重復數據的明細結果；

(2)然后對實際生產過程中的無效數據進行刪除。這里利用FlinkDataStream API的Evictor()方法對WindowFunction前后的數據進行處理。Evictor()方法包括Count-Evictor、DeltaEvictor和TimeEvictor以及自定義的Evictor。CountEvictor是在窗口中設置保持的數據數量，如：evictor(CountEvictor.of(10000))，意思是窗口中最大的數據量為10 000，若大于10 000條，則剔除。在實際生產過程中也會產生很多已過時的無效數據，其不僅會影響平臺數據查詢的正確性，而且會增加平臺的資源消耗，進而影響執行效率，而Flink 自帶的TimeEvictor方法能將最新時間的數據篩選出來，去除過時的數據。其主要將當前窗口中最新元素的時間減去時間間隔，然后將小于該結果的數據全部剔除。DeltaEvictor方法通過定義DeltaFunction和指定threshold(閾值)，計算出窗口間數據的Delt大小，如果超過了閾值則將當前數據元素刪除，這樣可以去除那些因為機器故障或者外部原因產生的差別較大的無效數據。同時也可以根據用戶的需求自定義Evictor方法來去除那些無效數據；

(3)利用步驟(1)中distinct去重方式的Distinct-Accumulator 與CountAccumulator方法統計單條數據value值的數量，DistinctAccumulator()內部包含一個map結構，key包含的是一條數據的屬性值，而value則是屬性值出現的次數。若缺少的value值過多(大于50%)，直接刪除缺失數據的記錄。反之認定數據為有效數據；

(4)原數據經過預處理后得到新數據，將這些數據存儲至數據池中。

通過一系列的數據預處理，可以有效防止臟數據影響平臺的正常運行。

2.4 數據存儲模塊

工業大數據異構性較強，數據類型較為復雜，這些數據通常以不同形式存儲在不同的數據庫或者數據管理系統中，所以管理起來較為麻煩，因此企業急需一個平臺對數據進行統一管理。而此平臺的數據源主要分為兩類，一類是實時數據，另一類是工業批數據，為了方便管理使用，此平臺建立一個數據池來存儲數據。實時數據存放至InfluxDB數據庫中，設備狀態、設備離線事件、設備事件等信息，靜態系統數據存放至MySQL數據庫中，如：企業設備、企業名稱、地址等信息。

數據存儲的過程如下：首先平臺先判斷獲取到的數據的類型，若是工業批數據會利用SQL-query去取出數據連接的URL、用戶名和密碼，然后加載SQL-JDBC去連接實例，并執行查詢；若是工業實時數據會先加載NoSQL-query，然后讀取NoSQL連接類，讀取InfluxDB數據庫自帶配置文件，從而連接實例，并執行查詢。

2.5 可視化分析模塊

無論是聯機事務處理(OLTP)，還是聯機分析處理(OLAP)，都是為了用戶更好地更直觀地獲取到處理到的數據結果，因此考慮一個與用戶交互性好的前端工具是十分必要的。

本平臺采用開源的Grafana作為可視化分析工具，它不僅支持多種數據庫，如IoTDB、MySQL、InfluxDB等，還支持多種數據的展示方式，如折線圖、圖表等，以更直觀的形式顯示出數據，用戶按照各自需求可快速獲取到數據且不需要關心后臺的具體運行過程。同時可以對工業設備進行預警，它通過Slack、DingDing、Email等方式通知企業數據已達到闕值，從而實現設備數據的準確預警。

2.6 集成過程

首先，數據源模塊可采用Flume收集工業生產過程中產生的日志，或直接從傳感器中獲取到數據，并由專門人員將這些數據整理為Excel格式。其次，利用Flink將整理好的數據發送至Kafka消息隊列中，保證數據傳輸的安全性和順序性。然后利用Flink獲取到暫存至Kafka中的數據并對其進行預處理，去除重復數據、缺失率較高的數據、無效數據等，處理好后將其存儲至不同的數據庫中，批數據存儲至MySQL中，流數據存儲至InfluxDB中。而Flink貫穿整個運行過程，對于MySQL中的數據可采用DataSet API，InfluxDB中的數據采用DataStream API。可視化分析模塊使用Grafana組件，實現不同類型數據的增刪改查，同時也可以對企業數據進行監測，若大于預定的值可通過郵件的形式進行預警。

3 實驗與分析

3.1 實驗環境和數據集

此平臺集群的硬件環境包含3臺物理機，一個為主節點，其余兩個為子節點，其域名分別為Master、Slave1、Slave2，3臺機器均使用8 G內存以及1 T的硬盤,使用的操作系統為Centos6.4 64位。Flink集群選擇1.9.3版本。Flink的master進程 jobManager放在Slave1中。修改好的配置文件放置在其它節點,并在Slave2的Flink_HOME/conf/slaves目錄下添加 Master、Slave1、Slvae2，這樣可以通過主節點免密登錄啟動其它的副節點啟動。Kafka應注意與Zookeeper 版本之間的兼容性，所以此平臺選擇了Kafka 2.2.0和Zookeeper3.4.10。前端工具Grafana選擇版本Grafana-6.7.2，此平臺選用的數據庫為MySQL-5.5和InfluxDB-1.7.3。

本文的實驗數據來自經過數據脫敏后的3000家企業基本信息，10 000余臺設備連續3個月的運行數據，脫敏簡要過程如下：利用Java代碼定義數據脫敏的工具類，涉及到具體公司名時，用*替代。同時企業id、設備id、事件id值重新編號，從而保護數據的安全性。本實驗的采用的數據主要分為兩種類型，一種是“企業名單”、“企業設備信息”、“測點名稱”等批數據，此類數據量為87 000條，描述的是3000家企業的一些基本信息，如公司名、所處地區等；另一種是“設備實時事件統計”、“設備狀態時長統計”等實時數據，此類數據量較大為1 250 000條，描述的是設備的實時狀態信息，如在某個時間段企業的狀態等。批數據以“企業名單”為例，包括5個字段值：企業id、客戶名稱、地址名稱、省市區、公司名，部分數據見表1。(注：表中只展示了部分數據集中的某個表，并不是全部數據)。

表1 企業名單

實時數據以“設備實時事件”數據為例，包括4個字段值：事件id、發生時間、設備id、事件，部分數據見表2。

表2 設備實時事件

3.2 平臺實現

為了驗證基于Flink的大數據平臺的有效性，對此平臺的各個模塊進行了測試。

系統實現具體過程如下：首先將數據源中的批數據和實時數據導入到Kafka消息隊列中。批數據的數據量較小，導入時間較短，耗時3 min 10 s成功將“企業名單”的信息發送至Kafka中。而“設備實時事件統計”實時數據量較大，耗時較長，耗時17 min成功將實時數據導入到Kafka中。然后利用Flink讀取Kafka數據并進行預處理后寫入到MySQL與InfluxDB中，預處理后的重復數據篩選結果見表3(以實時數據“設備實時事件”為例)。

表3 重復數據篩選結果

無效數據篩選結果見表4。

表4 無效數據篩選結果

空數據篩選結果見表5。

表5 空數據篩選結果

數據篩選后，開發人員可利用Navicat和InfluxDBStudio可視化工具查看數據，用戶顯示界面如圖4所示。

圖4 用戶界面

用戶在瀏覽器中輸入localhost:3000進入此平臺，首先填寫數據庫的用戶名以及密碼，創建用戶需要的數據庫，其次選擇數據庫顯示的儀表形式，有折線圖、表格、文本等形式。例如：用戶想查詢MySQL數據庫中的某個特定條件的批數據并以表格的形式輸出，用戶可在系統界面選擇Table并輸入:SELECT * FROM ′company_list′ WHERE 區=′天寧區′，便只查詢天寧區的公司名單，其實現效果如圖5所示。

圖5 公司名稱

同理，也可以實現實時數據的查詢，能快速地查詢到各個設備的實時狀態和對應的發生時間，如圖6所示。

圖6 設備實時狀態表

同時系統界面也提供edit的方式，用戶只需要選擇操作的數據庫和限定條件，也可輕松查詢到數據。不僅如此，用戶也可以利用此平臺篩選出自己所需要的數據，如查詢到相同設備號id的機器、同一時間內機器的上線數量、統計一段時間內出故障機器的數量、顯示預警的極值和結束時間等。

Grafana可以無縫定義告警在數據中的位置，可視化的定義閾值，并可以通過釘釘、E-mail等平臺獲取告警通知。這里我們選用E-mail的形式來關注實時設備狀態并獲得告警通知。首先在啟動Grafana前配置/etc/grafana/grafana.ini開啟smtp服務，配置發送郵件的郵箱以及密碼。配置好后，通過Grafana 的Alerting功能設置發送郵件的間隔時間，實現對設備數據的預警(此平臺判斷機器是否出現故障的方式有兩種：①由于機器是24小時運作的，所以機器會一直呈現在線狀態，若機器離線時間過長則會判定為出故障；②平臺會每隔一定時間發送機器上下線的數量給用戶，若下線的機器數量過多，則判定有機器出現故障)，其告警如圖7所示。

圖7 告警

3.3 分析比較

在文獻[11]中，作者介紹了當前較為典型的Clou-dera大數據平臺，其以Hadoop技術架構為基礎，具有穩定的、可擴展的企業級大數據管理平臺，它提供了很多部署案例，能夠方便管理企業生產過程中的多種數據，且具有強大的管理和監控工具。其中Cloudera Manager是開源的方便使用的一款產品，它提供Web用戶界面使得企業進行數據管理時更加容易。而Shark[12]也是一個相對較新的開源工業大數據分析平臺，它是Spark的一個組件，可安裝在與Hadoop相同的集群上，是一個性能較好的分布式和容錯內存分析系統，它具有數據聯合分區，容錯以及機器學習的能力，且完全兼容Hive和HiveQL，也能支持多種數據庫數據的查詢。

本平臺采用了Flink框架來構建工業大數據平臺。首先比較Flink平臺和文獻[11，12]二者平臺基礎框架的技術特點：Spark和Flink都是運行在YARN上的，但Flink的性能是優于Spark的，而Spark性能是大于Hadoop的，而且迭代的次數越多，Flink 的優勢越明顯。不僅如此Flink具有靈活的窗口，對于流數據處理起來更加方便，而工業生產下流數據偏多且較為復雜，因此Flink十分適用于工業場景。

其次，文獻[11，12]對于工業領域中的不同類型的數據無明確的區分，只采用單一的數據庫存儲數據。而本平臺中采用一個數據池來存儲不同數據，批數據放入MySQL據庫中，流數據放入InfluxDB數據庫中，能夠更好地區分開不同類型的工業大數據。再者，本平臺利用Kafka進行數據暫時存儲，更好地保證了數據傳輸的安全性以及平臺的可擴展性。

Flink平臺和Cloudera Manager大數據平臺、Shark大數據平臺的查詢數據效率如圖8所示，行表示數據集的數量(單位個數)，列表示用戶查詢數據的響應時間(單位ms)。

圖8 數據查詢效率

從圖8中我們可以看出：當數據集為5000條時，各個平臺的執行效率是差不多的，基本能在幾毫秒內響應出來。但當數據集數據變多時，Shark平臺和Cloudera Manager平臺數據查詢時間明顯上升，執行效率變低，而Flink平臺在處理將近60 000條數據集時也能快速響應。

3個平臺的吞吐量方面也進行了比較(吞吐量即單位時間內平臺成功傳送數據的數量)，比較結果如圖9所示，本次測試吞吐量的單位為：條/s。

圖9 吞吐量比較

從圖9中可以看到，當Kafka Data的Partition為1 時，此平臺的吞吐量是Cloudera Manager大數據平臺的3.2倍，是Shark平臺的將近1倍，而當Partition數為8時，此平臺吞吐量為Shark平臺的將近1倍，是Cloudera Manager大數據平臺的4.6倍。總之Flink平臺的吞吐量是遠遠高于其它兩個平臺的，而吞吐量又極大地反應了系統的負載能力。在工業大數據量大的情況下，Flink平臺能夠更好地運作。

當數據量變大時，延遲低也是一個企業需要考慮的地方，因此比較了3個平臺的延遲性。延遲性即數據從進入系統到流出系統所用的時間，本次測試延遲的單位為：ms。其實驗結果如圖10所示。

圖10 延遲比較

從圖10可以看到，Flink平臺的延遲較低，即使面對200 000條的數據量，平臺也只具有21 ms的延遲，而Shark平臺的延遲幾乎是Flink平臺的2倍，而Cloudera Manager平臺是Shark平臺的兩倍，因此Flink平臺在延遲上也有較大的優勢。

同時，在平臺預警速度方面做了個對比：選用5000條實時數據在不同的平臺上運行，比較不同平臺進行預警并發送郵件至用戶的時間。其實驗結果如圖11所示。

圖11 預警時間比較

從圖11中我們可以看出：5000條實時數據在此平臺進行預警并發送郵件的速度是最快的，需要20 ms，而在Shark平臺和Cloudera Manager大數據平臺分別需要27 ms和35 ms，此工業大數據平臺預警時間更短，能夠最大地減少企業的損失。

4 結束語

針對工業大數據數據量大、異構性強、及時性強的特點，引入大數據技術，提出了Flink和Kafka集成的工業大數據平臺，此平臺通過集群環境能夠高效地查詢數據，并能進行設備數據的快速預警。與目前較為典型的兩款開源大數據平臺進行比較，實驗結果表明，此平臺在數據查詢效率、吞吐量、延遲性以及預警速度方面都是優于其它兩個典型的大數據平臺的，能夠滿足預計的設計目標。不僅如此，此平臺不僅適用于工業領域，而且適用于所有時間序列數據多的場景，因此基于Flink的工業大數據平臺的研究是具有實際意義的。

在今后的工作中，還需完善此平臺的其它功能，如云平臺數據分析故障預測等。其次數據源部分的數據都是整理好的，而此平臺中并未過多介紹如何獲取數據源，因此今后還需多學習物聯網的知識。在企業生產過程中，安全性是重中之重的，雖然在傳輸過程中使用消息隊列保證數據傳輸的安全性，但其它模塊產生的數據實際是不夠安全的，所以如何保障保證數據處理的安全性[13]，這也是本平臺未來需要考慮的地方。