Kettle 4 2原始碼分析

乙個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的實現類在完成相應功能時，返回結果的型別。

主要成員變數：

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原始碼分析

Kettle 4 2原始碼分析第四講

Kettle 4 2原始碼分析第四講

4 2 list原始碼分析

Kettle 4 2原始碼分析

Kettle 4 2原始碼分析第四講

Kettle 4 2原始碼分析第四講

4 2 list原始碼分析

相關推薦