一、概述

在實際私有化物聯(lián)網(wǎng)平臺項目中，部分存量設備由于異構總線、多制式以太網(wǎng)、協(xié)議多樣化等因素導致無法直接連接物聯(lián)網(wǎng)平臺，大量數(shù)據(jù)較難集成，平臺側(cè)和設備側(cè)面臨大量定制化開發(fā)，成本較高。因此難以推動客戶或設備廠商進行存量設備接入改造，導致設備無法直連物聯(lián)網(wǎng)平臺，無法達到物聯(lián)網(wǎng)平臺對企業(yè)所有設備數(shù)據(jù)進行統(tǒng)一納管。

企業(yè)內(nèi)部存量的數(shù)據(jù)采集系統(tǒng)多為“煙囪式”，各個廠商的系統(tǒng)只需對接自己廠商的設備即可，數(shù)據(jù)孤島問題突出。

各“煙囪”的數(shù)據(jù)格式各不相同，定制化采集任務代碼不可復用，費時費力，難以同時支撐多個項目。

除了設備數(shù)據(jù)采集外，還有業(yè)務數(shù)據(jù)采集需求，傳統(tǒng)物聯(lián)網(wǎng)系統(tǒng)只能采集設備數(shù)據(jù)而無法集成業(yè)務數(shù)據(jù)。

二、技術選型

數(shù)字集成技術通過對不同系統(tǒng)數(shù)據(jù)的抽取(Extract)，數(shù)據(jù)清洗和轉(zhuǎn)換(Transformation)以及輸入最終的目標系統(tǒng)(Load)，打通各個業(yè)務孤島，實現(xiàn)數(shù)據(jù)互聯(lián)互通，助力企業(yè)數(shù)字化轉(zhuǎn)型。由于物聯(lián)網(wǎng)場景下的數(shù)據(jù)處理大多都要求實時性，所以要求實現(xiàn)時具備實時數(shù)據(jù)處理能力。實時計算也被稱作流計算，代表是Storm、Spark Streaming、Flink等大數(shù)據(jù)技術。計算引擎也在不斷更新迭代，從第一代的Hadoop MapReduce，到第二代的Spark，再到第三代的Flink技術，從批處理到微批，再到真正的流式計算。

Apache Flink是一個開源的流處理框架，應用于分布式、高性能、高可用的數(shù)據(jù)流應用程序。可以處理有限數(shù)據(jù)流和無限數(shù)據(jù)，即能夠處理有邊界和無邊界的數(shù)據(jù)流。無邊界的數(shù)據(jù)流就是真正意義上的流數(shù)據(jù)，所以Flink是支持流計算的。Flink可以部署在各種集群環(huán)境，可以對各種大小規(guī)模的數(shù)據(jù)進行快速計算。

Flink框架具備強大的流式ETL的能力，依靠其豐富的算子實現(xiàn)。

2.1 Source算子

Flink可以使用StreamExecutionEnvironment.addSource(source)來為我們的程序添加數(shù)據(jù)來源。

Flink已經(jīng)提供了若干實現(xiàn)好的source functions，當然也可通過實現(xiàn)SourceFunction來自定義非并行的source或者實現(xiàn)ParallelSourceFunction接口或者擴展RichParallelSourceFunction來自定義并行的source。

Flink在流處理上的source大致有4大類：

• 基于本地集合的source(Collection-based-source)
• 基于文件的source(File-based-source)- 讀取文本文件，即符合TextInputFormat規(guī)范的文件，并將其作為字符串返回
• 基于網(wǎng)絡套接字的source（Socket-based-source）- 從socket讀取。元素可以用分隔符切分。
• 自定義的source（Custom-source）

使用自定義Source算子可實現(xiàn)豐富的數(shù)據(jù)抽取功能。

2.2 Transform轉(zhuǎn)換算子

① map

將DataStream中的每一個元素轉(zhuǎn)換為另外一個元素，如將元素x變?yōu)樵瓉淼?倍：

dataStream.map { x => x * 5 }

② FlatMap

采用一個數(shù)據(jù)元并生成零個，一個或多個數(shù)據(jù)元。如，將句子分割為單詞的flatmap函數(shù)：

dataStream.flatMap { str => str.split("  ") }

③ Filter

計算每個數(shù)據(jù)元的布爾函數(shù)，并保存函數(shù)返回true的數(shù)據(jù)元。如，過濾掉零值的過濾器：

dataStream.filter { x != 0 }

當然flink還具備很多其他功能的轉(zhuǎn)換算子，如KeyBy、Reduce、Aggregations等，通過豐富的轉(zhuǎn)換算子，flink可實現(xiàn)對數(shù)據(jù)的清洗和轉(zhuǎn)換功能。

2.3 Sink算子

Flink的sink算子支持將數(shù)據(jù)輸出到：本地文件、本地集合、HDFS，除此之外，還支持：sink到kafka、sink到mysql、sink到redis以及自定義sink算子。

通過自定義sink算子將清洗轉(zhuǎn)換完成的數(shù)據(jù)輸入目標系統(tǒng)。

三、數(shù)字集成實現(xiàn)

實現(xiàn)過程如下：

第一步，抽象定義基礎控件類

數(shù)字集成基于flink可抽象定義3類基礎功能控件，每類控件又可根據(jù)不同的功能實現(xiàn)具體的子類功能控件；詳細如下：

基礎功能控件分為三類：數(shù)據(jù)源控件、數(shù)據(jù)輸出控件、數(shù)據(jù)處理控件。

數(shù)據(jù)源控件：將Source算子抽象定義成具備抽取數(shù)據(jù)功能的數(shù)據(jù)源控件類，并制定相應的配置規(guī)范，使用時只需根據(jù)規(guī)范配置文件，系統(tǒng)根據(jù)配置文件創(chuàng)建具體的實例化對象，實現(xiàn)數(shù)據(jù)抽取功能；

數(shù)據(jù)操作控件：根據(jù)不同的基礎功能需求將Transform算子抽象成數(shù)據(jù)處理控件類，制定相應的配置規(guī)范，使用時只需根據(jù)規(guī)范配置文件，系統(tǒng)根據(jù)配置創(chuàng)建相應的實例化對象實現(xiàn)數(shù)據(jù)處理功能；

數(shù)據(jù)輸出控件：將Sink算子抽象成數(shù)據(jù)輸出控件類，制定相應的配置規(guī)范，使用時只需根據(jù)規(guī)范配置文件，系統(tǒng)根據(jù)配置創(chuàng)建實例化對象實現(xiàn)數(shù)據(jù)輸出功能。

同時系統(tǒng)內(nèi)部明確定義flink算子之間流轉(zhuǎn)的數(shù)據(jù)格式作為內(nèi)部流轉(zhuǎn)數(shù)據(jù)格式以及根據(jù)配置輸出每個基礎功能控件輸出的數(shù)據(jù)格式。

第二步，根據(jù)抽象定義的基礎功能控件，制定具體配置規(guī)范

基礎功能控件規(guī)范如下：

通過以上兩步規(guī)范定義后，在同一個系統(tǒng)中，同一個處理過程只需要定義一個基礎功能控件規(guī)范。如Kafka消費者所需的配置如Kafka集群地址、消費群組、數(shù)據(jù)所在topic、數(shù)據(jù)所在分區(qū)key，消費位置等，只需要規(guī)定上述舉例這樣一個Kafka消費控件并開發(fā)實現(xiàn)，該控件就可以在該系統(tǒng)中復用，每次配置的數(shù)據(jù)處理工作流，復用Kafka消費控件類并根據(jù)新配置的源系統(tǒng)提供的Kafka集群地址、數(shù)據(jù)所在topic等配置即可實例化該工作流所需的kafka 消費者，實現(xiàn)過程從開發(fā)無數(shù)次Kafka Consumer的代碼變?yōu)閷崿F(xiàn)一次Kafka Consumer控件代碼，大量節(jié)省開發(fā)時間和開發(fā)成本。

第三步，通過對基本功能的抽象，實現(xiàn)如HTTP請求、kafka生產(chǎn)、數(shù)據(jù)遍歷、條件循環(huán)、數(shù)據(jù)映射、MySQL寫操作等基礎功能控件并實現(xiàn)，再根據(jù)各個基礎功能運行的先后邏輯組裝相應配置執(zhí)行腳本來編排組建成一個完整flink流處理鏈路，即可完成不同系統(tǒng)間的數(shù)據(jù)集成功能。

如在私有化項目中有將設備廠商云平臺中智能門鎖狀態(tài)信息同步至自有云平臺進行智能門鎖控制的需求，由于智能門鎖設備協(xié)議與自有物聯(lián)網(wǎng)平臺數(shù)據(jù)采集協(xié)議不適配，無法直連，由設備廠商云平臺提供智能門鎖狀態(tài)信息推送功能，由自有物聯(lián)網(wǎng)平臺提供推送數(shù)據(jù)接收接口，完成智能門鎖狀態(tài)信息的同步功能。

在此案例中，通過flink框架的自定義Source算子實現(xiàn)HTTP POST功能接口的HTTP監(jiān)聽控件完成設備廠商云平臺的推送數(shù)據(jù)接收功能，將接收到的智能門鎖狀態(tài)信息根據(jù)智能門鎖ID、狀態(tài)status與自有云平臺存儲的狀態(tài)進行比較的IF分支控件，將存在狀態(tài)變化的智能門鎖狀態(tài)信息數(shù)據(jù)向后序Sink算子流轉(zhuǎn)，通過自定義Sink算子實現(xiàn)自有云平臺數(shù)據(jù)上傳功能，完成智能門鎖狀態(tài)信息的跨平臺更新功能。

第四步，根據(jù)組建好的執(zhí)行邏輯生成有向無環(huán)圖，提交Flink運行，具體如下：

通過對不同的基礎功能控件，基于有向無環(huán)圖，將基礎功能控件放入有向無環(huán)圖的頂點，其中整個圖中只有一個數(shù)據(jù)源控件，且無其他基礎功能控件可以將數(shù)據(jù)傳輸給它；數(shù)據(jù)輸出控件和數(shù)據(jù)操作控件可以多個，對應多條分支處理邏輯。將數(shù)據(jù)傳輸方向作為有向無環(huán)圖的邊，以此連接和組織跨系統(tǒng)數(shù)據(jù)傳輸過程中針對數(shù)據(jù)的不同邏輯順序，生成一條完整的數(shù)據(jù)傳輸處理鏈路，將此圖完整實現(xiàn)，提交flink執(zhí)行，即可實現(xiàn)完整的數(shù)據(jù)抽取、轉(zhuǎn)換以及輸出的數(shù)字集成功能。

四、總結

最后我們來總結下基于Flink的數(shù)字集成能力的實現(xiàn)。得益于flink在ETL數(shù)據(jù)集成上的豐富能力以及算子之間易于處理的基礎功能，我們將flink的3類算子進行抽象定義實現(xiàn)3類基礎功能控件，實現(xiàn)不同的數(shù)據(jù)處理過程。根據(jù)不同的功能需求，通過Source算子實現(xiàn)從消息隊列、API、數(shù)據(jù)庫等多種數(shù)據(jù)源抽取數(shù)據(jù)的功能；通過豐富的Transform算子實現(xiàn)數(shù)據(jù)的清洗、篩選、轉(zhuǎn)換的功能；最后可通過Sink算子實現(xiàn)將目標格式數(shù)據(jù)輸入目標系統(tǒng)接收數(shù)據(jù)的渠道如消息隊列、數(shù)據(jù)庫、API等。綜上所述，基于Flink的數(shù)字集成能力是可以實現(xiàn)并且具備豐富功能和可擴展性的。