于起超 韓旭 馬丹璇 羅登昌

摘? 要： 為了解決傳統(tǒng)數(shù)據(jù)清洗工具面對(duì)海量數(shù)據(jù)時(shí)復(fù)雜度高、效率低的問題，設(shè)計(jì)實(shí)現(xiàn)了流式大數(shù)據(jù)數(shù)據(jù)清洗系統(tǒng)。利用分布式計(jì)算技術(shù)清洗數(shù)據(jù)，以解決性能低的問題。該系統(tǒng)由統(tǒng)一接入模塊、計(jì)算集群和調(diào)度中心三部分組成，實(shí)現(xiàn)了多種數(shù)據(jù)源的統(tǒng)一接入，分布式處理，并通過Web界面進(jìn)行清洗流程的交互式配置。實(shí)驗(yàn)結(jié)果表明，面對(duì)海量數(shù)據(jù)的時(shí)候，流式大數(shù)據(jù)數(shù)據(jù)清洗系統(tǒng)的性能強(qiáng)于傳統(tǒng)的單機(jī)數(shù)據(jù)清洗，提高了清洗效率。

關(guān)鍵詞： 數(shù)據(jù)清洗; 大數(shù)據(jù); 流式; 分布式架構(gòu)

中圖分類號(hào)：TP311? ? ? ? ? 文獻(xiàn)標(biāo)識(shí)碼：A? ? 文章編號(hào)：1006-8228（2021）09-01-04

Abstract： A streaming big data ETL system was designed and implemented so that the problem of high complexity and low efficiency of traditional ETL tools in the face of big data can be resolved. The system uses distributed computing technology to clean the data to solve the problem of low performance. The system consists of the unified access module， computing cluster and dispatching center. It realizes the unified access of multiple data sources and distributed processing， and the interactive configuration of cleaning process through web interface. The experiment results show that in the face of big data， the performance of the streaming big data ETL system is better than the traditional single machine ETL ones， which improves the cleaning efficiency.

Key words： ETL; big data; streaming; distributed architecture

0 引言

大數(shù)據(jù)的來臨改變了很多傳統(tǒng)工作方式，其中就包括ETL。ETL即數(shù)據(jù)抽?。‥xtract）、轉(zhuǎn)換（Transform）、清洗（Cleaning）、裝載（Loading）的過程，是數(shù)據(jù)倉(cāng)庫(kù)建設(shè)的重要環(huán)節(jié)，負(fù)責(zé)整個(gè)數(shù)據(jù)倉(cāng)庫(kù)的調(diào)度[1]。其效率的高低和清洗數(shù)據(jù)質(zhì)量的高低，直接決定數(shù)據(jù)倉(cāng)庫(kù)建設(shè)和決策的正確性。傳統(tǒng)的數(shù)據(jù)清洗框架有如下幾個(gè)問題。

⑴ 數(shù)據(jù)清洗性能低

傳統(tǒng)的數(shù)據(jù)清洗方式處理的數(shù)據(jù)量往往很小。在大數(shù)據(jù)時(shí)代數(shù)據(jù)量很可能每天成TB級(jí)別增加，加上有些清洗算法消耗很高的計(jì)算能力，導(dǎo)致傳統(tǒng)的數(shù)據(jù)清洗方法異常緩慢，甚至很難正常運(yùn)行[2]。

⑵ 數(shù)據(jù)源多樣性

大數(shù)據(jù)的多樣性導(dǎo)致數(shù)據(jù)源的復(fù)雜多樣[3]。數(shù)據(jù)倉(cāng)庫(kù)的數(shù)據(jù)往往來源于許多不同的系統(tǒng)，每個(gè)系統(tǒng)又包含多個(gè)模塊，每個(gè)模塊又包含單獨(dú)的數(shù)據(jù)源。從數(shù)據(jù)結(jié)構(gòu)來看，不僅包括結(jié)構(gòu)化數(shù)據(jù)還包括各種復(fù)雜的半結(jié)構(gòu)化數(shù)據(jù)[4]，這不僅提高了ETL程序的復(fù)雜性，也加大了維護(hù)的難度[5]。

針對(duì)現(xiàn)有數(shù)據(jù)清洗面臨的問題文中提出了流式大數(shù)據(jù)數(shù)據(jù)清洗系統(tǒng)，系統(tǒng)采用Kafka做中間件將接入數(shù)據(jù)與處理數(shù)據(jù)進(jìn)行解耦[6]，在一定程度上解決了傳統(tǒng)數(shù)據(jù)清洗框架普遍存在的問題。

1 系統(tǒng)設(shè)計(jì)

1.1 系統(tǒng)總體架構(gòu)設(shè)計(jì)

流式數(shù)據(jù)清洗架構(gòu)如圖1所示。系統(tǒng)涉及多種數(shù)據(jù)源，包括Excel、監(jiān)測(cè)日志、關(guān)系型數(shù)據(jù)庫(kù)等。數(shù)據(jù)源通過統(tǒng)一數(shù)據(jù)接入模塊進(jìn)行統(tǒng)一封裝后推送入分布式消息隊(duì)列Kafka中。計(jì)算集群消費(fèi)數(shù)據(jù)并執(zhí)行清洗操作，最終將清洗后的結(jié)果輸出到數(shù)據(jù)倉(cāng)庫(kù)中。

這種架構(gòu)主要有以下優(yōu)勢(shì)：①將不同類型數(shù)據(jù)都轉(zhuǎn)換成流的形式，使不同數(shù)據(jù)在形式上進(jìn)行統(tǒng)一。清洗數(shù)據(jù)的計(jì)算節(jié)點(diǎn)只需關(guān)心具體數(shù)據(jù)不需要處理數(shù)據(jù)來源問題。②清洗數(shù)據(jù)采用并行分布式方式處理，提高了數(shù)據(jù)清洗的性能。計(jì)算節(jié)點(diǎn)可以根據(jù)實(shí)際負(fù)載情況進(jìn)行擴(kuò)展，具有很強(qiáng)的擴(kuò)展性。③交互式的調(diào)度中心可以根據(jù)需求對(duì)清洗流程進(jìn)行可視化配置，降低了數(shù)據(jù)清洗的復(fù)雜度。

1.2 統(tǒng)一數(shù)據(jù)接入模塊設(shè)計(jì)

1.2.1 統(tǒng)一數(shù)據(jù)接入架構(gòu)設(shè)計(jì)

如圖2為統(tǒng)一數(shù)據(jù)接入的架構(gòu)圖，統(tǒng)一數(shù)據(jù)接入模塊主要包括定時(shí)器，文件監(jiān)控，SQL執(zhí)行三大子模塊。

⑴ 定時(shí)器

定時(shí)器模塊為文件監(jiān)控模塊和SQL執(zhí)行模塊提供定時(shí)功能，通過定時(shí)來控制數(shù)據(jù)的采集速率。用戶通過界面配置定時(shí)，可針對(duì)于每一種數(shù)據(jù)源定制執(zhí)行周期。

⑵ 文件監(jiān)控

文件監(jiān)控模塊是針對(duì)于日志文件的采集而設(shè)計(jì)的模塊。當(dāng)監(jiān)控的文件夾內(nèi)有文件增加時(shí)，文件監(jiān)控模塊讀取新增文件，將文件按照約定的解析規(guī)則進(jìn)行解析，生成規(guī)定的統(tǒng)一數(shù)據(jù)協(xié)議并推送kafka中。

⑶ SQL執(zhí)行

SQL執(zhí)行模塊實(shí)現(xiàn)對(duì)Mysql，Oracle，SQL Server等關(guān)系型數(shù)據(jù)庫(kù)的采集。SQL執(zhí)行模塊定時(shí)從數(shù)據(jù)庫(kù)中讀取一批次的數(shù)據(jù)，并轉(zhuǎn)化為統(tǒng)一數(shù)據(jù)協(xié)議推送Kafka中。

1.2.2 統(tǒng)一數(shù)據(jù)協(xié)議設(shè)計(jì)

統(tǒng)一數(shù)據(jù)協(xié)議如表1所示。主要有以下幾個(gè)字段。uuid為動(dòng)態(tài)生成的每條數(shù)據(jù)的唯一id，nameId為數(shù)據(jù)源的唯一id，timestamp為生產(chǎn)這條數(shù)據(jù)的時(shí)間。fields為一個(gè)字符串?dāng)?shù)組，存放關(guān)系型數(shù)據(jù)庫(kù)的字段名稱或者Excel的列名。dates存放具體的數(shù)據(jù)值。

1.3 計(jì)算集群模塊設(shè)計(jì)

計(jì)算集群由多個(gè)計(jì)算節(jié)點(diǎn)組成，計(jì)算節(jié)點(diǎn)架構(gòu)圖如圖3所示。

⑴ 接口

接口模塊用于和調(diào)度中心模塊和統(tǒng)一數(shù)據(jù)接入模塊進(jìn)行通信，包括數(shù)據(jù)源配置接口、集群管理接口、流程調(diào)度接口等接口。接口模塊采用RPC接口協(xié)議，RPC協(xié)議即Remote Produce Call遠(yuǎn)程過程調(diào)用協(xié)議，是一個(gè)計(jì)算機(jī)通信協(xié)議。該協(xié)議允許運(yùn)行于一臺(tái)計(jì)算機(jī)的程序調(diào)用另一臺(tái)計(jì)算機(jī)的子程序，而程序員無需額外地為這個(gè)交互作用編程。RPC基于高效的二進(jìn)制傳輸，與HTTP協(xié)議相比，字節(jié)大小和序列化耗時(shí)更低。

⑵ 同步器

同步器模塊用來與數(shù)據(jù)庫(kù)中調(diào)度作業(yè)進(jìn)行同步。該模塊保存作業(yè)運(yùn)行的實(shí)時(shí)狀態(tài)，并在作業(yè)重啟后讀取作業(yè)最后運(yùn)行狀態(tài)，從而保證清洗作業(yè)的正確運(yùn)行。

⑶ 元數(shù)據(jù)

元數(shù)據(jù)模塊保存數(shù)據(jù)源數(shù)據(jù)結(jié)構(gòu)的信息，緩存清洗數(shù)據(jù)的字典碼表信息。

⑷ 流程解析器

流程解析器模塊通過接口模塊讀取作業(yè)清洗流程的配置信息，并將配置信息解析成數(shù)據(jù)清洗對(duì)應(yīng)的有向無環(huán)圖。

⑸ 算子執(zhí)行器

算子執(zhí)行器讀取配置的清洗參數(shù)，并調(diào)用算子中的清洗方法進(jìn)行清洗。算子執(zhí)行器無需關(guān)心具體的清洗流程，只需關(guān)注于算子中的清洗方法，使得數(shù)據(jù)清洗具有可擴(kuò)展性。

1.4 調(diào)度中心模塊設(shè)計(jì)

調(diào)度中心模塊作為系統(tǒng)用戶交互的窗口，為用戶提供可視化的清洗流程配置界面，方便各種復(fù)雜清洗里的配置。調(diào)度中心模塊有數(shù)據(jù)源管理、集群配置、算子管理、清洗字典管理、清洗流程管理等功能模塊，如圖4所示。

⑴ 數(shù)據(jù)源管理

數(shù)據(jù)源管理模塊針對(duì)不同的數(shù)據(jù)源提供統(tǒng)一的配置管理功能。該模塊對(duì)數(shù)據(jù)源提供接入規(guī)則，主要包括定時(shí)周期、監(jiān)控文件夾、抽取SQL語句、統(tǒng)一數(shù)據(jù)協(xié)議等，并與統(tǒng)一數(shù)據(jù)接入模塊進(jìn)行交互，控制統(tǒng)一數(shù)據(jù)接入模塊的啟動(dòng)與停止。

⑵ 集群管理

集群管理模塊對(duì)統(tǒng)一接入模塊集群和計(jì)算集群提供管理和監(jiān)控功能，如監(jiān)控集群的上線和下線、監(jiān)控集群資源的利用情況、監(jiān)控清洗作業(yè)執(zhí)行情況并對(duì)錯(cuò)誤作業(yè)提供預(yù)警功能。

⑶ 算子管理

算子管理模塊提供對(duì)計(jì)算算子進(jìn)行統(tǒng)一管理功能。計(jì)算算子分為計(jì)算算子和輸出算子。計(jì)算算子用于數(shù)據(jù)清洗，輸出算子用于清洗結(jié)果的輸出。常用的輸出算子有Elasticsearch輸出算子、Hive輸出算子、數(shù)據(jù)庫(kù)輸出算子、Kafka輸出算子等。在添加算子的時(shí)候需要配置算子的執(zhí)行函數(shù)、算子描述、參數(shù)名、參數(shù)類型等。

⑷ 清洗算子字典管理

清洗算子字典管理模塊針對(duì)于字典替換算子而設(shè)計(jì)的。該模塊提供字典配置功能，并將映射關(guān)系緩存Redis中。字典替換算子讀取Redis緩存，并作字典映射處理。

⑸ 清洗流程管理

清洗流程管理模塊為用戶提供交互式的清洗流程配置功能。用戶通過Web界面，在畫布上拖拽清洗算子，配置每一個(gè)清洗算子對(duì)應(yīng)的清洗參數(shù)，并按照清洗規(guī)則進(jìn)行連線，形成一個(gè)由起點(diǎn)到輸出的流程圖。這種可視化的配置方式，給用戶提供了直觀的清理流程控制，降低了數(shù)據(jù)清洗的復(fù)雜度。

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

2.1 計(jì)算節(jié)點(diǎn)數(shù)據(jù)清洗的實(shí)現(xiàn)

計(jì)算節(jié)點(diǎn)的清洗流程如圖5所示。計(jì)算節(jié)點(diǎn)主動(dòng)從Kafka拉取消息，并獲取消息中的nameId，判斷nameId是否存在清洗規(guī)則庫(kù)，若清洗規(guī)則庫(kù)不存在則跳過清洗流程，若存在則進(jìn)行清洗環(huán)節(jié)。計(jì)算節(jié)點(diǎn)從調(diào)度中心臨時(shí)庫(kù)獲取對(duì)應(yīng)的清洗流程圖，并遍歷流程圖中的每一個(gè)算子。算子利用JAVA的反射原理調(diào)用清洗算子中的清洗函數(shù)執(zhí)行清洗工作。

2.2 計(jì)算算子的實(shí)現(xiàn)

計(jì)算算子將清洗規(guī)則進(jìn)行封裝形成一個(gè)完整的可控的工具類，主要包括數(shù)據(jù)驗(yàn)證，數(shù)據(jù)轉(zhuǎn)換，數(shù)據(jù)去重，缺失插值等[7]。

本文計(jì)算算子的設(shè)計(jì)參考了Hive的UDF的設(shè)計(jì)理論，UDF（User Defined Function）提供了一種接口，可以對(duì)流中的每一條消息進(jìn)行相應(yīng)的處理[8]。算子類繼承接口UDF，并實(shí)現(xiàn)接口evaluate的函數(shù)，evaluate函數(shù)有兩個(gè)參數(shù)一個(gè)是從流中獲取的一條消息protocol，args是從調(diào)度中心獲取的自定義參數(shù)。

3 實(shí)驗(yàn)分析對(duì)比

3.1 實(shí)驗(yàn)環(huán)境

為了對(duì)比傳統(tǒng)方式和基于流式計(jì)算分布式清洗框架數(shù)據(jù)清洗的性能，本實(shí)驗(yàn)采用三臺(tái)服務(wù)器來部署計(jì)算集群，用一臺(tái)服務(wù)器部署傳統(tǒng)清洗方式Kettle。具體的配置如表2所示。

實(shí)驗(yàn)數(shù)據(jù)來自與地質(zhì)的險(xiǎn)情日志數(shù)據(jù)，現(xiàn)采用100萬、300萬、500萬與800萬條日志數(shù)據(jù)進(jìn)行實(shí)驗(yàn)測(cè)試。測(cè)試計(jì)算算子有字典替換算子，輸出算子是Elasticsearch批量輸出算子。

3.2 實(shí)驗(yàn)結(jié)果

如圖6所示為傳統(tǒng)Kettle單機(jī)方式和分布式數(shù)據(jù)清洗方式所用時(shí)間的折線圖。通過折線圖可以看出，隨著數(shù)據(jù)量增加，分布式方式斜率變緩，且所用時(shí)間明顯小于Kettle的傳統(tǒng)方式。

4 結(jié)束語

隨著大數(shù)據(jù)技術(shù)的不斷發(fā)展，數(shù)據(jù)清洗在異構(gòu)數(shù)據(jù)集成和數(shù)據(jù)倉(cāng)庫(kù)等領(lǐng)域的研究與應(yīng)用越來越引起人們的重視[9]。但由于數(shù)據(jù)的復(fù)雜性，數(shù)據(jù)清洗具有很大的難度[10]。本文從數(shù)據(jù)清洗的概念出發(fā)，分析了大數(shù)據(jù)背景下數(shù)據(jù)清洗所面臨的困難與問題。提出了一種基于流式計(jì)算的數(shù)據(jù)清洗框架，并對(duì)框架架構(gòu)和各個(gè)模塊進(jìn)行剖析，并通過實(shí)驗(yàn)對(duì)比分析，證明了系統(tǒng)在運(yùn)行效率有良好的表現(xiàn)?？梢钥闯鱿到y(tǒng)具有很強(qiáng)的擴(kuò)展性，通用性，大大簡(jiǎn)化了數(shù)據(jù)清洗的操作流程，具有一定的實(shí)際價(jià)值[11]。此框架待完善的方面有：①數(shù)據(jù)流程圖目前只能支持簡(jiǎn)單的流程配置，更為復(fù)雜的流程還未適配到調(diào)度中心。②算子的擴(kuò)展性有待提高，目前支持的算子有限。③支持的數(shù)據(jù)源有限，目前只支持三種數(shù)據(jù)類型，未來可增加更多的數(shù)據(jù)種類。④計(jì)算集群計(jì)算能力有待提高，未來可將計(jì)算集群移植到Spark，Hadoop等并行計(jì)算框架上[12]。

參考文獻(xiàn)（References）：

[1] 劉佳俊，喻鋼，胡珉.面向城市基礎(chǔ)設(shè)施智慧管養(yǎng)的大數(shù)據(jù)智能融合方法[J].計(jì)算機(jī)應(yīng)用，2017.37（10）：2983-2990

[2] 何剛.基于Hadoop平臺(tái)的分布式ETL研究與實(shí)現(xiàn)[D].東華大學(xué)，2014.

[3] 黃毅，鐘碧良.基于XML的異構(gòu)數(shù)據(jù)庫(kù)間數(shù)據(jù)遷移的研究[J].科技管理研究，2008.28（8）：173-174

[4] 劉華，胡燕，王濤.Web數(shù)據(jù)清洗研究[J].軟件導(dǎo)刊，2007.3：75-77

[5] 葉舟，王東.基于規(guī)則引擎的數(shù)據(jù)清洗[J].計(jì)算機(jī)工程，2006.32（23）：52-54

[6] Garg N.Learning Apache Kafka-Second Edition[J].2015.

[7] Kandel S， Heer J， Plaisant C， et al. Research directions in data wrangling：visualizations and transformations for usable and credible data[J]. Information Visualization，2016.10（4）：271-288

[8] 范會(huì)麗，彭寧，任薇.基于Hadoop平臺(tái)的數(shù)據(jù)清洗研究[J].電腦知識(shí)與技術(shù)，2020.16（5）：33-34

[9] 李垚周，李光明.分布式數(shù)據(jù)清洗系統(tǒng)設(shè)計(jì)[J].網(wǎng)絡(luò)安全技術(shù)與應(yīng)用，2020.2.

[10] Rahm E， Do Honghai. Data Cleaning： Problems and? Current Approaches[J].Data Engineering，2000.23（4）.

[11] 王銘軍，潘巧明，劉真，et al.可視數(shù)據(jù)清洗綜述[J]. 中國(guó)圖象圖形學(xué)報(bào)，2015.20（4）：468-482

[12] Li X， Mao Y. Real-Time data ETL framework for big real-time data analysis[C]//IEEE International Conference on Information & Automation.IEEE，2015.10.