Labs 導(dǎo)讀

數(shù)據(jù)集成平臺(tái)作為連接各種異構(gòu)數(shù)據(jù)的紐帶，需要連接多種多樣的存儲(chǔ)系統(tǒng)。而不同的技術(shù)棧和不同的業(yè)務(wù)場(chǎng)景會(huì)對(duì)數(shù)據(jù)集成系統(tǒng)提出不同的設(shè)計(jì)要求。

1

概述

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

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

各“煙囪”的數(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ù)而無(wú)法集成業(yè)務(wù)數(shù)據(jù)。

2

技術(shù)選型

數(shù)字集成技術(shù)通過(guò)對(duì)不同系統(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)場(chǎ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è)開(kāi)源的流處理框架，應(yīng)用于分布式、高性能、高可用的數(shù)據(jù)流應(yīng)用程序?？梢蕴幚碛邢迶?shù)據(jù)流和無(wú)限數(shù)據(jù)，即能夠處理有邊界和無(wú)邊界的數(shù)據(jù)流。無(wú)邊界的數(shù)據(jù)流就是真正意義上的流數(shù)據(jù)，所以Flink是支持流計(jì)算的。Flink可以部署在各種集群環(huán)境，可以對(duì)各種大小規(guī)模的數(shù)據(jù)進(jìn)行快速計(jì)算。

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

2.1 Source算子

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

Flink已經(jīng)提供了若干實(shí)現(xiàn)好的source functions，當(dāng)然也可通過(guò)實(shí)現(xiàn)SourceFunction來(lái)自定義非并行的source或者實(shí)現(xiàn)ParallelSourceFunction接口或者擴(kuò)展RichParallelSourceFunction來(lái)自定義并行的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)樵瓉?lái)的5倍：

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ù)元。如，過(guò)濾掉零值的過(guò)濾器：

dataStream.filter { x != 0 }

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

2.3 Sink算子

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

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

3

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

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

第一步，抽象定義基礎(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í)例化對(duì)象，實(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í)例化對(duì)象實(shí)現(xiàn)數(shù)據(jù)處理功能；

數(shù)據(jù)輸出控件：將Sink算子抽象成數(shù)據(jù)輸出控件類，制定相應(yīng)的配置規(guī)范，使用時(shí)只需根據(jù)規(guī)范配置文件，系統(tǒng)根據(jù)配置創(chuàng)建實(shí)例化對(duì)象實(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ò)以上兩步規(guī)范定義后，在同一個(gè)系統(tǒng)中，同一個(gè)處理過(guò)程只需要定義一個(gè)基礎(chǔ)功能控件規(guī)范。如Kafka消費(fèi)者所需的配置如Kafka集群地址、消費(fèi)群組、數(shù)據(jù)所在topic、數(shù)據(jù)所在分區(qū)key，消費(fèi)位置等，只需要規(guī)定上述舉例這樣一個(gè)Kafka消費(fèi)控件并開(kā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)過(guò)程從開(kāi)發(fā)無(wú)數(shù)次Kafka Consumer的代碼變?yōu)閷?shí)現(xiàn)一次Kafka Consumer控件代碼，大量節(jié)省開(kāi)發(fā)時(shí)間和開(kāi)發(fā)成本。

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

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

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

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

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

4

總結(jié)

最后我們來(lái)總結(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ù)處理過(guò)程。根據(jù)不同的功能需求，通過(guò)Source算子實(shí)現(xiàn)從消息隊(duì)列、API、數(shù)據(jù)庫(kù)等多種數(shù)據(jù)源抽取數(shù)據(jù)的功能；通過(guò)豐富的Transform算子實(shí)現(xiàn)數(shù)據(jù)的清洗、篩選、轉(zhuǎn)換的功能；最后可通過(guò)Sink算子實(shí)現(xiàn)將目標(biāo)格式數(shù)據(jù)輸入目標(biāo)系統(tǒng)接收數(shù)據(jù)的渠道如消息隊(duì)列、數(shù)據(jù)庫(kù)、API等。綜上所述，基于Flink的數(shù)字集成能力是可以實(shí)現(xiàn)并且具備豐富功能和可擴(kuò)展性的。

審核編輯：劉清