一、概述

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

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

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

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

二、技術(shù)選型

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

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

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

2.1 Source算子

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

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

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

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

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

2.2 Transform轉(zhuǎn)換算子

① map

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

dataStream.map { x => x * 5 }

② FlatMap

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

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

③ Filter

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

dataStream.filter { x != 0 }

當(dāng)然flink還具備很多其他功能的轉(zhuǎn)換算子，如KeyBy、Reduce、Aggregations等，通過豐富的轉(zhuǎn)換算子，flink可實(shí)現(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ù)輸入目標(biāo)系統(tǒng)。

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

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

第一步，抽象定義基礎(chǔ)控件類

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

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

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

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

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

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

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

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

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

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

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

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

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

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

四、總結(jié)

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