于起超 韓旭 馬丹璇 羅登昌

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

關鍵詞： 數(shù)據清洗; 大數(shù)據; 流式; 分布式架構

中圖分類號：TP311? ? ? ? ? 文獻標識碼：A? ? 文章編號：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ù)據的來臨改變了很多傳統(tǒng)工作方式，其中就包括ETL。ETL即數(shù)據抽?。‥xtract）、轉換（Transform）、清洗（Cleaning）、裝載（Loading）的過程，是數(shù)據倉庫建設的重要環(huán)節(jié)，負責整個數(shù)據倉庫的調度[1]。其效率的高低和清洗數(shù)據質量的高低，直接決定數(shù)據倉庫建設和決策的正確性。傳統(tǒng)的數(shù)據清洗框架有如下幾個問題。

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

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

⑵ 數(shù)據源多樣性

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

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

1 系統(tǒng)設計

1.1 系統(tǒng)總體架構設計

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

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

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

1.2.1 統(tǒng)一數(shù)據接入架構設計

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

⑴ 定時器

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

⑵ 文件監(jiān)控

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

⑶ SQL執(zhí)行

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

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

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

1.3 計算集群模塊設計

計算集群由多個計算節(jié)點組成，計算節(jié)點架構圖如圖3所示。

⑴ 接口

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

⑵ 同步器

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

⑶ 元數(shù)據

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

⑷ 流程解析器

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

⑸ 算子執(zhí)行器

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

1.4 調度中心模塊設計

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

⑴ 數(shù)據源管理

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

⑵ 集群管理

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

⑶ 算子管理

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

⑷ 清洗算子字典管理

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

⑸ 清洗流程管理

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

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

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

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

2.2 計算算子的實現(xiàn)

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

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

3 實驗分析對比

3.1 實驗環(huán)境

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

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

3.2 實驗結果

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

4 結束語

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

參考文獻（References）：

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

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

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

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

[5] 葉舟，王東.基于規(guī)則引擎的數(shù)據清洗[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] 范會麗，彭寧，任薇.基于Hadoop平臺的數(shù)據清洗研究[J].電腦知識與技術，2020.16（5）：33-34

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

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

[11] 王銘軍，潘巧明，劉真，et al.可視數(shù)據清洗綜述[J]. 中國圖象圖形學報，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.