Kettle 4 2原始碼分析

2021-06-26 14:30:51 字數 3923 閱讀 1988



乙個job項代表etl控制流中的一項邏輯任務。job項將會順序執行,每個job項會產生乙個結果,能作為別的分支上job項的條件。

圖 1 job項示例

圖 2 job entry類圖結構

jobentryinte***ce是job entry外掛程式的主要實現介面。主要包含以下功能:

1          儲存job entry設定

實現類使用私有變數儲存設定的引數,通過get、set方法獲取和設定。dialog實現類會通過這些方法,儲存或設定設定介面上的引數。同時,需要提供乙個深度拷貝的方法,因為在一些儲存引數且可能修改的地方會呼叫。

圖 3 jobentrytrans配置介面

2       序列化外掛程式

外掛程式要實現對本外掛程式的序列化,實現兩種方式xml與資料庫。

圖 4 轉換外掛程式xml序列化結果

3          輸出資訊提供

乙個job entry支援三種型別的輸出:true、flase和無條件。這三種情況不是所有的job entry外掛程式都會同時支援的,例如dummy job entry僅支援true和false。所以,外掛程式必須顯現兩類函式,來檢視支援哪種結果。

public boolean evaluates()//是否支援true、false

public boolean isunconditional()//是否支援無條件執行

4          執行任務

負責工作的執行。

public result execute()//執行具體的邏輯,需要結果和開始到該項的距離

prev_result.setnrerrors()//

設定執行過程中的異常數

prev_result.setresult()//

設定結果,如果不知道true/false,結果不設定

最後返回prev_result。

負責構建和開啟引數設定對話方塊。spoon通過呼叫open函式開啟該對話方塊,spoon是使用swt框架的,所以對話方塊也應使用swt來實現。

每乙個jobentryinte***ce的實現類在完成相應功能時,返回結果的型別。

主要成員變數:

1

private

boolean

result;執行是否出現異常23

private

intexitstatus; 執行結果狀態45

private listrows;乙個jobentry完成處理後的資料(若存在)67

圖 5 job開啟時序圖

job的開啟與trans相類似,配置執行的引數,檢查.kjb檔案是否發生變化,例項化乙個job物件,開啟該執行緒。

主要工作是從jobmeta的jobhopmeta找到job入口jobentry資訊,根據開始條件呼叫真正執行jobentry的execute方法2,**如下所示:

** 4 job.excute()關鍵**

1 startpoint=jobmeta.findjobentry(jobmeta.string_special_start, 0,    false);//

找到job開始元件

2 jobentryspecial jes =(jobentryspecial) startpoint.getentry();3//

jobentryspecial是啟動job的job專案

4 result res = null;5

while ( (jes.isrepeat() || isfirst) && !isstopped())

execute()方法包含,的引數有執行次數(start不算,從0開始,順序執行)、接乙個entry執行結果、當前entry的拷貝、前乙個entry拷貝和原因。

主要功能是根據引數startpoint,提取對應的jobentry,執行對應的jobentry操作,再根據jobmeta的hop資訊依次得到下乙個jobentry,遞迴呼叫。具體的執行步驟如下所示:

圖 6 job執行步驟

具體每個元件的執行體對應org.pentaho.di.job.entries包內每個entry的具體實現。

execute()方法2中呼叫jobentry的execute()完成jobentry的具體功能。

finalresult result = clonejei.execute(prevresult, nr, rep,this);

不同的job專案(jobentry)實現差別很大。

jobentryspecial

功能是開啟乙個job,只是簡單地對傳遞來的preresult設定它的的result屬性值為true,(job專案據此判斷前一結果執行完畢)。返回該物件即可。

jobentrytableexit

功能是判斷乙個table是否存在資料庫中。jobentrytableexit job專案有屬性tablename和databasemeta(對資料庫的元資料資訊描述)根據databasemeta得到乙個dabase物件db,建立連線db.connect(); 呼叫db.checktableexists(tablename)根據此返回值設定preresult的result屬性為否為true。返回preresult物件。

jobentrytrans

jobentryjob和jobentrytrans是巢狀job或trans的job專案(jobentry)。它們是比較複雜的job專案。

作用是執行乙個trans。首先例項化乙個transmeta,之後例項化trans。呼叫trans.start(),當執行完畢後呼叫函式trans.getresult(),並把結果加到preresult中,返回該物件即可。

補充說明

result中也可以有處理資料,這些處理資料可以作為下乙個job專案(jobentry)的輸入。但是容量受記憶體容量限制。

pdi使用資料庫外掛程式來進行資料庫的正確連線、執行sql,同時也考慮現有資料的各種特殊功能和不同限制。

在pdi裡面,已經整合了非常多的資料庫外掛程式,大部分的外掛程式都會繼承自basedatabasemeta。下面所示的方法通常都需要被重寫,基類裡面並沒有相關的實現。要實現的方法主要分成3大主題:連線資訊、sql方言和功能標記。

1連線詳情

當pdi建立資料庫連線時將會呼叫這些函式,或者資料庫設定對話方塊裡顯示與方言有關的內容時也會呼叫。

2sql generation

構建有效的sql資料庫方言時會呼叫這些方法。

3功能標記

查詢使用的資料庫是否支援該功能。

ps:kettle原始碼分析算是全部講完了,最後奉送自己做的ppt 應該算是比較小眾的軟體,但是在業界還是非常有名氣的。我看到過好幾個中國的公司,包括上市公司,說自己最新的etl工具或者資料共享交換工具都是kettle的改版。

Kettle 4 2原始碼分析第四講

乙個job項代表etl控制流中的一項邏輯任務。job項將會順序執行,每個job項會產生乙個結果,能作為別的分支上job項的條件。圖 1 job項示例 圖 2 job entry類圖結構 jobentryinte ce是job entry外掛程式的主要實現介面。主要包含以下功能 1儲存job entr...

Kettle 4 2原始碼分析第四講

乙個job項代表etl控制流中的一項邏輯任務。job項將會順序執行,每個job項會產生乙個結果,能作為別的分支上job項的條件。圖 1 job項示例 圖 2 job entry類圖結構 jobentryinte ce是job entry外掛程式的主要實現介面。主要包含以下功能 1儲存job entr...

4 2 list原始碼分析

list概述 list底層為非連續區間,即鍊錶 實質上是乙個雙向迴圈鍊錶 list每次插入或者刪除乙個元素,就配置和釋放乙個元素空間,對於任何位置的原屬插入或原屬移除,list永遠為常數時間。list的節點 首先要知道,list本身和list的節點是不同的,如果我們宣告乙個list,裡面放了100w...