流水線處理技術(shù)在數(shù)據(jù)集成中的應(yīng)用

隨著個人計算機(jī)和計算機(jī)網(wǎng)絡(luò)的飛速發(fā)展，以及信息化的高速推進(jìn)，互聯(lián)網(wǎng)提供的信息總量也在迅猛增長。如果企業(yè)和社會組織實現(xiàn)數(shù)據(jù)共享，可以使更多的人更充分地利用已有的數(shù)據(jù)資源?？墒菫椴煌瑧?yīng)用服務(wù)的信息都存儲在許多不同的數(shù)據(jù)源之中，數(shù)據(jù)內(nèi)容以及數(shù)據(jù)格式千差萬別，且其管理系統(tǒng)也各不相同。如何對這些數(shù)據(jù)進(jìn)行有效的集成管理，屏蔽這些信息的異構(gòu)，并提供一個統(tǒng)一的訪問接口以透明地訪問各信息源，成為一些大型企業(yè)或社會組織關(guān)心的事情。數(shù)據(jù)集成正是在這一背景下提出的。

1 基于數(shù)據(jù)復(fù)制方法的集成模式

數(shù)據(jù)復(fù)制方法[1]是當(dāng)前比較常用的數(shù)據(jù)集成模式，該方法將各個數(shù)據(jù)源的數(shù)據(jù)復(fù)制到與其相關(guān)的其他數(shù)據(jù)源上，并維護(hù)數(shù)據(jù)源整體上的數(shù)據(jù)一致性、提高信息共享利用的效率。這種方式可以復(fù)制信息源的整個數(shù)據(jù)，也可以是信息源的部分信息。數(shù)據(jù)復(fù)制方法在用戶使用某個數(shù)據(jù)源之前，將用戶可能用到的其他數(shù)據(jù)源的數(shù)據(jù)預(yù)先復(fù)制過來，如果用戶要使用的數(shù)據(jù)已經(jīng)被復(fù)制，則只需要查詢該集成信息源，并與中介器/包裝器的虛擬數(shù)據(jù)集成[2]相比，大大提高了系統(tǒng)處理用戶請求的效率。

基于數(shù)據(jù)復(fù)制方式最常見的一種方法是數(shù)據(jù)倉庫方法[1]。該方法將各個數(shù)據(jù)的全部或者部分?jǐn)?shù)據(jù)復(fù)制到數(shù)據(jù)倉庫，用戶像訪問普通數(shù)據(jù)庫一樣直接訪問數(shù)據(jù)倉庫。該方式實現(xiàn)了對物理數(shù)據(jù)庫語義異構(gòu)的屏蔽和數(shù)據(jù)訪問的控制，提供了一個統(tǒng)一的數(shù)據(jù)邏輯視圖來隱藏底層的數(shù)據(jù)細(xì)節(jié)。圖1所示為一個典型的數(shù)據(jù)倉庫體系結(jié)構(gòu)圖[3]。

在該集成模型中，每一個數(shù)據(jù)源對應(yīng)一個監(jiān)視器（Monitor），監(jiān)視器負(fù)責(zé)收集所需要集成的信息源中數(shù)據(jù)的變化以便上報給集成端(收集的方式有如下類別：針對信息源有日志的情況，可以通過日志分析提取要上報的增量；對于沒有日志情況可以通過觸發(fā)器方式或者快照差分方式獲取信息源的增量)，同時監(jiān)視器還具有一個包裝器的功能，提供信息源的數(shù)據(jù)查詢提取以及類型轉(zhuǎn)化功能。當(dāng)作為數(shù)據(jù)查詢功能的時候，不僅將數(shù)據(jù)初始化同步到數(shù)據(jù)倉庫中，同時也相當(dāng)于一個服務(wù)器，不斷偵聽來自于集成器的命令查詢請求，當(dāng)有請求到達(dá)時，執(zhí)行查詢，并將該監(jiān)視器對應(yīng)的數(shù)據(jù)源的數(shù)據(jù)包裝成基于公共類型的數(shù)據(jù)，或以XML文件的方式和固定大小對象數(shù)據(jù)塊的方式傳遞給集成器，然后集成器負(fù)責(zé)將提取后的數(shù)據(jù)進(jìn)行合成。其中監(jiān)視器與集成器中的通信流程如圖2所示。

2 基于內(nèi)存控制的流水線處理方法

從上面的數(shù)據(jù)倉庫體系結(jié)構(gòu)可知，監(jiān)視器必須具備一個包裝器（wrapper）的功能。數(shù)據(jù)倉庫端保存的數(shù)據(jù)是各底層信息源的部分副本（一般情況為訪問非常頻繁），但是不是很頻繁的訪問數(shù)據(jù)還是保持在底層信息源端，當(dāng)一個OLAP查詢（如下鉆）經(jīng)過查詢分解后，不能在數(shù)據(jù)倉庫端獲取，而必須通過一個甚至幾個底層信息源端的查詢，然后在集成器端進(jìn)行結(jié)果的合并（如要通過兩個底層數(shù)據(jù)庫中表的連接操作）才能獲取查詢結(jié)果。在實化視圖初始化時，提高查詢的效率以及提高實化視圖初始化的時間，是非常重要的。

本文關(guān)注的便是如何提高查詢效率、響應(yīng)速度、集成端的處理效率，以及在提交查詢后，如何減少集成端的空閑等待時間，并且在大數(shù)據(jù)量的情況下同時做到內(nèi)存控制，以防止在大數(shù)據(jù)量的情況下查詢導(dǎo)致內(nèi)存溢出。

在解決提高查詢效率、響應(yīng)速度、集成端的插入效率的同時，防止內(nèi)存溢出以及在減少集成端的空閑等待時間方面，利用了基于生產(chǎn)者/消費(fèi)者的流水線處理方法，該方式主要思想是實現(xiàn)服務(wù)器與客戶端的流水并行[4]，查詢的結(jié)果以固定大小數(shù)據(jù)塊的形式分塊組裝，并在監(jiān)視器端與集成端都使用一定大小的緩沖隊列來暫存這些數(shù)據(jù)塊，以有效防止內(nèi)存溢出。以一次實化視圖的初始化過程為例，描述該方式的算法流程為：

(1)集成端發(fā)送帶全局查詢QID（該查詢QID為全局唯一的，通過客戶端API自動生成）的SQL查詢命令(結(jié)果查詢重寫)，并通過通信平臺將該查詢命令放入服務(wù)器端執(zhí)行隊列中，同時預(yù)設(shè)一個數(shù)據(jù)塊計數(shù)為sum（該計數(shù)為服務(wù)器端初始要發(fā)送的數(shù)據(jù)塊個數(shù)），然后集成端監(jiān)聽接收隊列；

(2)監(jiān)視器端從命令隊列中取出查詢命令，創(chuàng)建查詢管理器（Data Query Manager），并將該查詢管理器與查詢QID作為一個鍵值對放入進(jìn)程全局的哈希表（Concurrent Hash Map)中，然后通過該查詢管理器中的excuteQurey()方法啟動查詢線程，該查詢線程將獲得的記錄組織成數(shù)據(jù)塊(Data Object Block)，放入固定大小的數(shù)據(jù)塊緩沖隊列中，并在該隊列滿時，查詢線程暫停，不滿時繼續(xù)查詢，直到最后一塊為止。同時啟動發(fā)送固定大小的數(shù)據(jù)塊的線程，該發(fā)送線程從緩沖隊列中取出數(shù)據(jù)塊，發(fā)送給客戶端，直到發(fā)送的最后一塊，該發(fā)送線程終止；

(3)當(dāng)有數(shù)據(jù)塊到達(dá)客戶端的數(shù)據(jù)塊接收隊列時，判斷該塊是否為最后一塊，如果是，則設(shè)置所有塊是否到達(dá)的標(biāo)志“flag=true”，并通知客戶端進(jìn)行處理，客戶端處理線程從隊列中取出一個數(shù)據(jù)塊進(jìn)行處理（對實化視圖初始化，處理的方式就是將該數(shù)據(jù)塊插入到數(shù)據(jù)倉庫的實化視圖中），并將數(shù)據(jù)塊計數(shù)n減1，再判斷該數(shù)據(jù)塊計數(shù)是否小于客戶端要緩沖的個數(shù)N，并同時判斷flag的值，如果sumN，且flag= =false，則發(fā)送從服務(wù)器端調(diào)取固定數(shù)目K數(shù)據(jù)塊的命令（該命令帶QID，以便到服務(wù)器端時找到之前的查詢管理器），同時設(shè)置sum=sum+K；

(4)服務(wù)器端接收到客戶端的數(shù)據(jù)塊調(diào)取命令，分離出里面的QID，從進(jìn)程全局的哈希表中找到與該QID對應(yīng)的查詢管理器，并調(diào)用里面的發(fā)送固定數(shù)據(jù)塊的方法以啟動發(fā)送固定數(shù)目數(shù)據(jù)塊的線程，該線程與步驟(2)中發(fā)送線程相同；

(5)重復(fù)步驟(3)、步驟(4)，直到查詢的最后一塊到達(dá)客戶端，與此同時，服務(wù)器端的查詢管理器也從全局的哈希表中移除。

3 性能測試與分析

與流水線處理方法相對應(yīng)的一種方法為同步方法，即通過查詢先將底層信息源的結(jié)果組裝在一起，一次傳給集成端處理。由于采用的都是對象數(shù)據(jù)塊的形式，因此用于與流水線對比的同步方法的算法思想為：

(1)客戶端發(fā)送帶全局查詢QID（該查詢QID為全局唯一的，通過客戶端API自動生成）的SQL查詢命令(結(jié)果查詢重寫)，并通過通信平臺將該查詢命令放入服務(wù)器端執(zhí)行隊列中；

(2)服務(wù)器端接收到查詢命令，執(zhí)行查詢，將所得的結(jié)果存放于文件中，然后一次發(fā)送給客戶端；

(3)客戶端接收到關(guān)于本次查詢結(jié)果集的文件，然后處理該結(jié)果集文件。

將基于內(nèi)存控制的流水線處理方法與同步方法在以下實驗環(huán)境下進(jìn)行測試對比，為減少誤差，多次測試得出平均值，有如下數(shù)據(jù)：

監(jiān)視器端與集成端采用相同配置環(huán)境，相關(guān)配置為：
CPU：Intel(R) Core(TM)2 Duo CPU E4500 @ 2.2 GHz；操作系統(tǒng)：Windows XP；內(nèi)存：2.0 GB；數(shù)據(jù)庫：Oracle 9i；JDK版本：1.6.0_07；開發(fā)工具：Myeclipse6.5。

本實驗性能測試如圖3所示，可以看出，與傳統(tǒng)的同步方法相比，采用本文算法具有較好的性能特性，主要在于基于內(nèi)存控制的流水線處理過程是一個監(jiān)視器端與集成器端并行流水線運(yùn)行的過程，并充分應(yīng)用了現(xiàn)在多處理器多線程處理的技術(shù)，減少了集成端空閑等待的時間。

設(shè)查詢信息源并包裝所有數(shù)據(jù)成公共類型數(shù)據(jù)塊的時間代價為Cost(Q)，傳輸放入文件中的所有數(shù)據(jù)塊到集成端的時間為Cost(T)，集成端將傳輸過來的數(shù)據(jù)解析并初始化到數(shù)據(jù)倉庫的時間為Cost(P)，則基于同步方法的時間代價為：Cost(Q)+Cost(T)+Cost(P)。

設(shè)查詢信息源并包裝查詢的數(shù)據(jù)成公共類型數(shù)據(jù)塊為一塊的時間代價為：Cost(Q1)，傳輸其中一塊數(shù)據(jù)塊到集成端的時間為Cost(T1)，集成端將傳輸過來的一塊數(shù)據(jù)塊解析并初始化到數(shù)據(jù)倉庫的時間為Cost(P1)，因為這里數(shù)據(jù)塊是個固定的常數(shù)，則基于本文的算法的時間代價為：Cost(Q1)+Cost(T1)+Cost(P1)+max(Cost(Q)-Cost(Q1)，Cost(T)-Cost(T1)，Cost(P)-Cost(P1))，其中max為各處理邏輯減去初始處理的最大時間。

從上面理論上可以分析得出，基于內(nèi)存控制的流水線處理技術(shù)較同步技術(shù)可以更好地提高效率。同時還存在幾個問題：

(1)當(dāng)集成端需要OLAP查詢或?qū)嵒晥D初始化比較多時，仍然會出現(xiàn)內(nèi)存溢出的問題，這時可以應(yīng)用線程池技術(shù)[4]，有效控制這類線程運(yùn)行的數(shù)量，同樣，監(jiān)視器端也使用這種方案。

(2)當(dāng)集成端與監(jiān)視器端進(jìn)行流水線處理時，如果監(jiān)視器端與集成端出現(xiàn)網(wǎng)絡(luò)中斷，或者其中一個出現(xiàn)突發(fā)事件（如斷電）時，之前的一些過程就需要重做，并回滾。特別是針對網(wǎng)絡(luò)中斷的情況，容易造成監(jiān)視器端查詢線程的線程泄漏，即集成端認(rèn)為之前的操作沒成功，然后重新進(jìn)行操作，然而監(jiān)視器端的處理線程卻還沒完。避免這些情況出現(xiàn)的解決方案為：設(shè)置一個超時，當(dāng)達(dá)到設(shè)定時間而這一流水處理過程未進(jìn)行時，自動中斷這些處理流程，或者可以在監(jiān)視器端對查詢組裝后數(shù)據(jù)塊分塊存儲在硬盤上，然后進(jìn)行文件數(shù)據(jù)塊的發(fā)送，這樣減少了塊之間的命令的交互邏輯，而且有效地控制了線程泄漏，但是也增加了文件的讀寫與控制，增加了I/O開銷。

數(shù)據(jù)集成仍然是一個比較熱門的研究點，在基于數(shù)據(jù)倉庫方法的數(shù)據(jù)集成中，分析了實化視圖初始化以及OLAP查詢中面對大數(shù)據(jù)量處理的問題，應(yīng)用了基于內(nèi)存控制的流水線處理方法，充分利用了Java的多線程處理技術(shù)，并從實驗和理論上分析了該方法較同步方法的優(yōu)點。