數據集成系統和數據集成方法

2023-09-18 15:18:50

數據集成系統和數據集成方法
【專利摘要】本發明提供了一種數據集成系統和數據集成方法，其中，該數據集成系統包括：日誌挖掘單元，用於通過第一線程對源資料庫進行日誌挖掘，將挖掘出的數據中符合第一預設條件的數據實例成對象，把對象存儲至緩存；處理單元，用於通過第二線程輪詢緩存，以從緩存中獲取符合第二預設條件的對象進行處理得到處理數據，將處理數據發送至消息隊列；寫入單元，用於通過第三線程輪詢消息隊列，並在消息隊列中存在消息時獲取消息，對消息進行處理得到目標語句，將語句寫入目標資料庫。通過本申請的技術方案，可以有效地提高數據集成的穩定性和數據集成的效率。
【專利說明】數據集成系統和數據集成方法
【技術領域】
[0001]本發明涉及數據處理【技術領域】，具體而言，涉及一種數據集成系統和一種數據集成方法。
【背景技術】
[0002]目前在雲計算環境下BI (商業智能)戰略變得越來越重要，而在商業智能中實時的數據集成在其中佔有很重要的部分，需要一種完成數據抽取，轉換，加載，分析等一整套數據處理流程的方法。
[0003]目前現有的實時數據集成的方法穩定性能比較差，對於列存儲的數據倉庫如gbase也沒有很好的支持，並且效率也不是很高，因此有必要對現有的增量實時數據集成進行優化。

【發明內容】

[0004]本發明正是基於上述問題，提出了一種數據集成技術，能夠有效地提高數據集成的穩定性和數據集成的效率。
[0005]有鑑於此，本發明提出了 一種數據集成系統，包括:日誌挖掘單元，用於通過第一線程對源資料庫進行日誌挖掘，將挖掘出的數據中符合第一預設條件的數據實例成對象，把所述對象存儲至緩存；處理單元，用於通過第二線程輪詢所述緩存，以從所述緩存中獲取符合第二預設條件的對象進行處理得到處理數據，將所述處理數據發送至消息隊列；寫入單元，用於通過第三線程輪詢所述消息隊列，並在所述消息隊列中存在消息時獲取所述消息，對所述消息進行處理得到目標語句，將所述語句寫入目標資料庫。
[0006]在上述技術方案中，優選地，還包括:持久化單元，用於對所述挖掘出的數據進行持久化。
[0007]在上述技術方案中，優選地，還包括:記錄單元，用於通過所述第二線程記錄所述處理數據的標識。
[0008]在上述技術方案中，優選地，所述寫入單元還用於通過所述第三線程將所述目標語句批量寫入所述目標資料庫。
[0009]在上述技術方案中，優選地，所述第一線程、所述第二線程和所述第三線程是異步執行的。
[0010]本申請還提出了一種數據集成方法，包括:步驟202，通過第一線程對源資料庫進行日誌挖掘，將挖掘出的數據中符合第一預設條件的數據實例成對象，把所述對象存儲至緩存；步驟204，通過第二線程輪詢所述緩存，以從所述緩存中獲取符合第二預設條件的對象進行處理得到處理數據，將所述處理數據發送至消息隊列；步驟206，通過第三線程輪詢所述消息隊列，並在所述消息隊列中存在消息時獲取所述消息，對所述消息進行處理得到目標語句，將所述語句寫入目標資料庫。
[0011]在上述技術方案中，優選地，所述步驟202還包括:對所述挖掘出的數據進行持久化。
[0012]在上述技術方案中，優選地，所述步驟204還包括:通過所述第二線程記錄所述處理數據的標識。
[0013]在上述技術方案中，優選地，所述步驟206還包括:通過所述第三線程將所述目標語句批量寫入所述目標資料庫。
[0014]在上述技術方案中，優選地，所述第一線程、所述第二線程和所述第三線程是異步執行的。
[0015]通過以上技術方案，可以有效地提高數據集成的穩定性和數據集成的效率。
【專利附圖】

【附圖說明】
[0016]圖1示出了根據本發明的實施例的數據集成系統的示意框圖；
[0017]圖2示出了根據本發明的實施例的數據集成方法的示意流程圖；
[0018]圖3示出了根據本發明的實施例的數據集成系統的結構示意圖；
[0019]圖4示出了根據本發明的實施例的數據流向示意圖；
[0020]圖5示出了根據本發明的實施例的數據挖掘和發送至消息隊列的示意流程；
[0021]圖6示出了根據本發明的實施例的數據寫入目標資料庫的示意流程圖。
【具體實施方式】
[0022]為了能夠更清楚地理解本發明的上述目的、特徵和優點，下面結合附圖和【具體實施方式】對本發明進行進一步的詳細描述。需要說明的是，在不衝突的情況下，本申請的實施例及實施例中的特徵可以相互組合。
[0023]在下面的描述中闡述了很多具體細節以便於充分理解本發明，但是，本發明還可以採用其他不同於在此描述的其他方式來實施，因此，本發明的保護範圍並不受下面公開的具體實施例的限制。
[0024]圖1示出了根據本發明的實施例的數據集成系統的示意框圖。
[0025]如圖1所示，根據本發明的實施例的數據集成系統100包括:日誌挖掘單元102，用於通過第一線程對源資料庫進行日誌挖掘，將挖掘出的數據中符合第一預設條件的數據實例成對象，把對象存儲至緩存；處理單元104，用於通過第二線程輪詢緩存，以從緩存中獲取符合第二預設條件的對象進行處理得到處理數據，將處理數據發送至消息隊列；寫入單元106，用於通過第三線程輪詢消息隊列，並在消息隊列中存在消息時獲取消息，對消息進行處理得到目標語句，將語句寫入目標資料庫。
[0026]從源端資料庫中挖掘出數據，並將數據寫入目標資料庫，需要先對源端資料庫的進行日誌挖掘，然後將挖掘出的數據放入緩存隊列，再將數據發送至消息隊列，最後寫入到目標資料庫，通過為日誌挖掘操作、將數據發送至消息隊列操作和寫入目標資料庫三步操作分別設置線程，可以使得上述三個操作同時進行，比如通過對源端資料庫進行日誌挖掘操作得到數據A，然後將數據A放入緩存隊列，在將緩存中的數據A發送至消息的隊列時，進行日誌挖掘操作的線程(即第一線程)可以繼續對源端數據進行日誌挖掘，並且將數據寫入目標資料庫的線程(即第三線程)也可以在此時將在數據A之前已經存在於消息隊列中的其他數據，繼續寫入目標資料庫，而無需等待數據A寫入目標資料庫之後才能再次對遠端資料庫進行日誌挖掘，也無需等待數據A寫入目標資料庫之後才將消息隊列中的其他數據寫入目標資料庫，從整體上提高了數據獲取、處理和寫入的效率，提高了實時數據集成的效率。
[0027]其中的第一預設條件和第二預設條件，可以預先設置一張數據表來規定所需的業務數據，對於即將實例化的數據和即將從緩存中發送至消息隊列的數據，可以判斷其是否輸入預先設置的表格中所規定的內容，若不屬於，說明該數據並不是用戶所需，則可以直接刪除，從而減少對無用數據的處理，提高了實時數據集成的效率。
[0028]優選地，還包括:持久化單元108，用於對挖掘出的數據進行持久化。
[0029]由於進行日誌挖掘得到的內容都存在於緩存中，如果進行日誌挖掘的線程(即第一線程)中斷或停止，那麼存在於緩存中的內容將全部丟失，進而導致進行需要同步的操作數據丟失，通過將挖掘出的數據進行持久化處理，比如持久化到本地日誌中，從而可以在進行日誌挖掘的線程中斷或停止的情況下，依然能夠通過查詢本地日誌來調用挖掘出的數據，提高了實時數據集成的穩定性。
[0030]優選地，還包括:記錄單元110，用於通過第二線程記錄處理數據的標識。
[0031]若第二線程發生中斷或停止，那麼挖掘出的數據會由於沒有發送到消息隊列中而丟失，通過記錄處理數據設置標識，具體可以是在每一個處理數據發送到消息隊列之前，都記錄下該數據當前的SCN號，從而避免第二線程崩潰而導致處理數據丟失，提高了實時數據集成的穩定性。
[0032]優選地，寫入單元106還用於通過第三線程將目標語句批量寫入目標資料庫。
[0033]可以在目標語句積累到一定數量後，通過批量導入的方式，將目標語句寫入目標資料庫，從而避免逐條寫入語句而造成的頻繁交互，提高了實時數據集成的效率。
[0034]優選地，還包括:所述第一線程、所述第二線程和所述第三線程是異步執行的。從而提聞每步操作的同步率，進而提聞實時數據集成的效率。
[0035]圖2示出了根據本發明的實施例的數據集成方法的示意流程圖。
[0036]如圖2所示，根據本發明的實施例的數據集成方法包括:步驟202，通過第一線程對源資料庫進行日誌挖掘，將挖掘出的數據中符合第一預設條件的數據實例成對象，把對象存儲至緩存；步驟204，通過第二線程輪詢緩存，以從緩存中獲取符合第二預設條件的對象進行處理得到處理數據，將處理數據發送至消息隊列；步驟206，通過第三線程輪詢消息隊列，並在消息隊列中存在消息時獲取消息，對消息進行處理得到目標語句，將語句寫入目標資料庫。
[0037]從源端資料庫中挖掘出數據，並將數據寫入目標資料庫，需要先對源端資料庫的進行日誌挖掘，然後將挖掘出的數據放入緩存隊列，再將數據發送至消息隊列，最後寫入到目標資料庫，通過為日誌挖掘操作、將數據發送至消息隊列操作和寫入目標資料庫三步操作分別設置線程，可以使得上述三個操作同時進行，比如通過對源端資料庫進行日誌挖掘操作得到數據A，然後將數據A放入緩存隊列，在將緩存中的數據A發送至消息的隊列時，進行日誌挖掘操作的線程(即第一線程)可以繼續對源端數據進行日誌挖掘，並且將數據寫入目標資料庫的線程(即第三線程)也可以在此時將在數據A之前已經存在於消息隊列中的其他數據，繼續寫入目標資料庫，而無需等待數據A寫入目標資料庫之後才能再次對遠端資料庫進行日誌挖掘，也無需等待數據A寫入目標資料庫之後才將消息隊列中的其他數據寫入目標資料庫，從整體上提高了數據獲取、處理和寫入的效率，提高了實時數據集成的效率。
[0038]在上述技術方案中，優選地，步驟202還包括:對挖掘出的數據進行持久化。
[0039]由於進行日誌挖掘得到的內容都存在於緩存中，如果進行日誌挖掘的線程(即第一線程)中斷或停止，那麼存在於緩存中的內容將全部丟失，進而導致進行需要同步的操作數據丟失，通過將挖掘出的數據進行持久化處理，比如持久化到本地日誌中，從而可以在進行日誌挖掘的線程中斷或停止的情況下，依然能夠通過查詢本地日誌來調用挖掘出的數據，提高了實時數據集成的穩定性。
[0040]在上述技術方案中，優選地，步驟204還包括:通過第二線程記錄處理數據的標識。
[0041]若第二線程發生中斷或停止，那麼挖掘出的數據會由於沒有發送到消息隊列中而丟失，通過記錄處理數據設置標識，具體可以是在每一個處理數據發送到消息隊列之前，都記錄下該數據當前的SCN號，從而避免第二線程崩潰而導致處理數據丟失，提高了實時數據集成的穩定性。
[0042]在上述技術方案中，優選地，步驟206還包括:通過第三線程將目標語句批量寫入目標資料庫。
[0043]可以在目標語句積累到一定數量後，通過批量導入的方式，將目標語句寫入目標資料庫，從而避免逐條寫入語句而造成的頻繁交互，提高了實時數據集成的效率。
[0044]在上述技術方案中，優選地，所述第一線程、所述第二線程和所述第三線程是異步執行的。從而提聞每步操作的同步率，進而提聞實時數據集成的效率。
[0045]其中的第一預設條件和第二預設條件，可以預先設置一張數據表來規定所需的業務數據，對於即將實例化的數據和即將從緩存中發送至消息隊列的數據，可以判斷其是否輸入預先設置的表格中所規定的內容，若不屬於，說明該數據並不是用戶所需，則可以直接刪除，從而減少對無用數據的處理，提高了實時數據集成的效率。
[0046]圖3示出了根據本發明的實施例的數據集成系統的結構示意圖。
[0047]如圖3所示，根據本申請的實施例的數據集成系統是基於日誌分析和消息中間件技術(數據集成，簡稱RDI)，內部具有高緩存，高並發的架構。可以通過具體有以下功能的軟體和/或硬體來實現:
[0048]Source DB:用於進行源端的日誌挖掘，相當於日誌挖掘單元；
[0049]RDI Source Agent:從源端資料庫分析日誌得到事務並發送至消息中間件，相當於處理單元；
[0050]RDI Admin Console:用來管理數據集成的實例中的任務和資源；
[0051]Message Queue (簡稱MQ):消息隊列中間件用於傳遞消息；
[0052]RDI Target Agent:從消息中間件取得事務並應用到目的端資料庫；
[0053]Target DB:用於寫入的目標資料庫，RDI Target Agent和Target DB相當於寫入單元。
[0054]其相應的數據流向如圖4所示，對源端資料庫402進行日誌挖掘得到的數據放入緩存隊列404，然後將緩存隊列404中的數據發送至消息隊列406，最後將消息隊列406中的數據寫入目標資料庫408。[0055]圖5示出了根據本發明的實施例的數據挖掘和發送至消息隊列的示意流程。
[0056]如圖5所示，數據集成引擎啟動時會開啟一個線程(相當於第一線程)專門用於進行日誌挖掘，挖掘的對象是資料庫的歸檔日誌，挖掘的工具使用的是oracle自帶的1gminer,主要是根據資料庫的系統變更號進行挖掘。線程啟動後首先會向1gminer裡傳遞一個開始系統變更號和一個結束系統變更號，1gminer會根據挖掘的範圍對歸檔日誌進行挖掘(當然，如果該挖掘操作發生在系統崩潰後，則可以根據記錄下的SCN號進行日誌挖掘)，挖掘出記錄後會選擇有效的欄位內容並實例成對象，最後把實例好的對象存入系統緩存中等待下一步處理。在這裡所有發掘出來的內容都存在於系統的緩存中因此如果發掘線程突然中斷或者停止，則存在與緩存中的內容都會丟失，帶來的問題就是進行需要同步的操作數據丟失，因此提出的改進方案就是把挖掘出來的數據持久化到本地的日誌中，這樣可以方便在出錯的時候對發掘的數據進行查詢。
[0057]數據集成引擎在啟動後會開啟另一個線程(相當於第二線程)專門用於把存在於系統緩存中的數據發送到消息隊列中。它與日誌挖掘的線程是異步執行的，這樣可以增加同步的效率。線程啟動後會輪詢系統的緩存隊列，如果發現有有效的實例化對象，會把該對象放入一個事務處理的對象中，直到一個事務提交完畢後，線程會把該事務對象送入配置好的過濾器進行過濾操作，沒有被過濾掉的數據會通過發送通道發送到MQ (即MessageQueue)消息隊列中，等待下一步處理。在這裡提出的改進方案是在每一個消息發送到隊列時都記錄下當前的SCN號以防止在系統崩潰時已發掘出來的但還沒有發送到消息隊列的數據丟失。
[0058]圖6示出了根據本發明的實施例的數據寫入目標資料庫的示意流程圖。
[0059]數據集成引擎在啟動後會開啟一個線程(相當於第三線程)用於獲取MQ消息隊列中的數據並把其寫入資料庫。線程啟動後會輪詢MQ消息隊列，如果發現有消息，會把消息中的事務對象提取出來，然後傳入事務攔截器進行處理，最後提取出處理好的事務對象中的SQL語句並寫入目標資料庫完成數據的同步過程。但目前數據導入的效率比較低，是因為數據採用的是逐行導入的方式，改進方案是提供一個批量導入的方法，這樣整體同步的效率會增加，寫入效率也會提升。
[0060]以上結合附圖詳細說明了本發明的技術方案，考慮到相關技術中，數據集成的方法穩定性能比較差，並且效率也不是很高。通過本申請的技術方案，能夠有效地提高數據集成的穩定性和數據集成的效率。
[0061]在本發明中，術語「第一」、「第二」僅用於描述目的，而不能理解為指示或暗示相對重要性。術語「多個」指兩個或兩個以上，除非另有明確的限定。
[0062]以上所述僅為本發明的優選實施例而已，並不用於限制本發明，對於本領域的技術人員來說，本發明可以有各種更改和變化。凡在本發明的精神和原則之內，所作的任何修改、等同替換、改進等，均應包含在本發明的保護範圍之內。
【權利要求】
1.一種數據集成系統，其特徵在於，包括: 日誌挖掘單元，用於通過第一線程對源資料庫進行日誌挖掘，將挖掘出的數據中符合第一預設條件的數據實例成對象，把所述對象存儲至緩存； 處理單元，用於通過第二線程輪詢所述緩存，以從所述緩存中獲取符合第二預設條件的對象進行處理得到處理數據，將所述處理數據發送至消息隊列； 寫入單元，用於通過第三線程輪詢所述消息隊列，並在所述消息隊列中存在消息時獲取所述消息，對所述消息進行處理得到目標語句，將所述語句寫入目標資料庫。
2.根據權利要求1所述的數據集成系統，其特徵在於，還包括: 持久化單元，用於對所述挖掘出的數據進行持久化。
3.根據權利要求1所述的數據集成系統，其特徵在於，還包括: 記錄單元，用於通過所述第二線程記錄所述處理數據的標識。
4.根據權利要求1所述的數據集成系統，其特徵在於，所述寫入單元還用於通過所述第三線程將所述目標語句批量寫入所述目標資料庫。
5.根據權利要求1至4中任一項所述的數據集成系統，其特徵在於，所述第一線程、所述第二線程和所述第三線程是異步執行的。
6.一種數據集成方法，其特徵在於，包括: 步驟202，通過第一線程對源資料庫進行日誌挖掘，將挖掘出的數據中符合第一預設條件的數據實例成對象，把所述對象存儲至緩存； 步驟204，通過第二線程輪詢所述緩存，以從所述緩存中獲取符合第二預設條件的對象進行處理得到處理數據，將所述處理數據發送至消息隊列； 步驟206，通過第三線程輪詢所述消息隊列，並在所述消息隊列中存在消息時獲取所述消息，對所述消息進行處理得到目標語句，將所述語句寫入目標資料庫。
7.根據權利要求6所述的數據集成方法，其特徵在於，所述步驟202還包括:對所述挖掘出的數據進行持久化。
8.根據權利要求6所述的數據集成方法，其特徵在於，所述步驟204還包括:通過所述第二線程記錄所述處理數據的標識。
9.根據權利要求6所述的數據集成方法，其特徵在於，所述步驟206還包括:通過所述第三線程將所述目標語句批量寫入所述目標資料庫。
10.根據權利要求6至9中任一項所述的數據集成方法，其特徵在於，所述第一線程、所述第二線程和所述第三線程是異步執行的。
【文檔編號】G06F17/30GK103488690SQ201310392454
【公開日】2014年1月1日 申請日期:2013年9月2日 優先權日:2013年9月2日
【發明者】衡駿 申請人:用友軟體股份有限公司