使用Kettle進行數據庫遷移操做

思路：先從源庫讀取須要遷移的表名，將代表做爲變量，並在目標庫建立這些表，最後表到表抽取數據

整個遷移流程包括2個job，4個transform。

具體以下：

1.總的Job:

2.獲取表名流程的轉換：

3.子job表數據抽取做業

3.1轉換 表名變量設置

 

3.2轉換 建立表結構

表輸入以下，注意：這裏的查詢必需要查詢出數據才行，否則後面會沒法獲取表結構來建立表

（有些人說這裏加個條件where 1=2，不須要數據只須要表結構，我本身測試，發現這樣沒法獲取到數據，後面的java腳本中能夠打印這些信息，加了條件會報錯，打印信息爲null）

java腳本：

public boolean processRow(StepMetaInterface smi, StepDataInterface sdi) throws KettleException
    {
    	// First, get a row from the default input hop
    	//
    	Object[] r = getRow();
		//logBasic("r.size="+r.length);
   		org.pentaho.di.core.database.DatabaseMeta dbmeta = null;
    	java.util.List list = getTrans().getRepository().readDatabases();//3.x中獲取資源庫的全部數據庫鏈接信息用getDatabases();
    	if(list != null && !list.isEmpty())
    	{
    		for(int i=0;i<list.size();i++)
    		{
    			dbmeta = (org.pentaho.di.core.database.DatabaseMeta)list.get(i);
    			//下面是目標庫的數據庫鏈接，你們可根據須要修改
    			if("testb".equalsIgnoreCase(dbmeta.getName()))
    			{
					logBasic("數據庫鏈接名爲:"+dbmeta.getName());
    				break;
    			}
    		}
    	}
    	if(dbmeta!=null)
    	{
    		org.pentaho.di.core.database.Database db=new org.pentaho.di.core.database.Database(dbmeta);
    		try
    		{
    			db.connect();
    			String tablename = getVariable("TABLENAME");
    			logBasic("開始建立表：" + tablename);
    			if(tablename!=null && tablename.trim().length()>0)
    			{
					logBasic("data.inputRowMeta="+data.inputRowMeta);
    				String sql = db.getDDL(tablename, data.inputRowMeta);//${TABLENAME}
					logBasic("sql="+sql);
    				db.execStatement(sql.replace(";", ""));
    				logBasic(sql);
    			}
    		}catch(Exception e){
    			logError("建立表出現異常",e);
    		}finally{
    			db.disconnect();
    		}
    	}
    	return false;
	}

3.3數據抽取

運行結果：

2017/07/24 15:16:25 - Spoon - 正在開始任務...
2017/07/24 15:16:25 - 數據庫遷移做業 - 開始執行任務
2017/07/24 15:16:25 - 數據庫遷移做業 - 開始項[獲取表名流程]
2017/07/24 15:16:25 - 源表名稱獲取 - 爲了轉換解除補丁開始  [源表名稱獲取]
2017/07/24 15:16:25 - 表輸入.0 - Finished reading query, closing connection.
2017/07/24 15:16:25 - 表輸入.0 - 完成處理 (I=2, O=0, R=0, W=2, U=0, E=0)
2017/07/24 15:16:25 - 字段選擇.0 - 完成處理 (I=0, O=0, R=2, W=2, U=0, E=0)
2017/07/24 15:16:25 - 複製記錄到結果.0 - 完成處理 (I=0, O=0, R=2, W=2, U=0, E=0)
2017/07/24 15:16:25 - 數據庫遷移做業 - 開始項[表數據抽取做業]
2017/07/24 15:16:25 - 表數據抽取子做業 - 開始項[代表成變量設置]
2017/07/24 15:16:25 - 將記錄中的表名設置爲變量 - 爲了轉換解除補丁開始  [將記錄中的表名設置爲變量]
2017/07/24 15:16:25 - 從結果獲取記錄.0 - 完成處理 (I=0, O=0, R=1, W=1, U=0, E=0)
2017/07/24 15:16:25 - 設置變量.0 - Setting environment variables...
2017/07/24 15:16:25 - 設置變量.0 - Set variable TABLENAME to value [TESTM]
2017/07/24 15:16:25 - 設置變量.0 - Finished after 1 rows.
2017/07/24 15:16:25 - 設置變量.0 - 完成處理 (I=0, O=0, R=1, W=1, U=0, E=0)
2017/07/24 15:16:25 - 表數據抽取子做業 - 開始項[建立表結構]
2017/07/24 15:16:25 - 建立表結構 - 爲了轉換解除補丁開始  [建立表結構]
2017/07/24 15:16:26 - 表輸入.0 - Finished reading query, closing connection.
2017/07/24 15:16:26 - 建立入庫表結構.0 - 數據庫鏈接名爲:testb
2017/07/24 15:16:26 - 表輸入.0 - 完成處理 (I=1, O=0, R=0, W=1, U=0, E=0)
2017/07/24 15:16:26 - 建立入庫表結構.0 - 開始建立表：TESTM
2017/07/24 15:16:26 - 建立入庫表結構.0 - data.inputRowMeta=[ID String(10)], [NAME String(200)], [AGE BigNumber], [ADDRESS String(200)]
2017/07/24 15:16:26 - 建立入庫表結構.0 - sql=CREATE TABLE TESTM
2017/07/24 15:16:26 - 建立入庫表結構.0 - (
2017/07/24 15:16:26 - 建立入庫表結構.0 -   ID VARCHAR2(10)
2017/07/24 15:16:26 - 建立入庫表結構.0 - , NAME VARCHAR2(200)
2017/07/24 15:16:26 - 建立入庫表結構.0 - , AGE NUMBER
2017/07/24 15:16:26 - 建立入庫表結構.0 - , ADDRESS VARCHAR2(200)
2017/07/24 15:16:26 - 建立入庫表結構.0 - )
2017/07/24 15:16:26 - 建立入庫表結構.0 - ;
2017/07/24 15:16:26 - 建立入庫表結構.0 - CREATE TABLE TESTM
2017/07/24 15:16:26 - 建立入庫表結構.0 - (
2017/07/24 15:16:26 - 建立入庫表結構.0 -   ID VARCHAR2(10)
2017/07/24 15:16:26 - 建立入庫表結構.0 - , NAME VARCHAR2(200)
2017/07/24 15:16:26 - 建立入庫表結構.0 - , AGE NUMBER
2017/07/24 15:16:26 - 建立入庫表結構.0 - , ADDRESS VARCHAR2(200)
2017/07/24 15:16:26 - 建立入庫表結構.0 - )
2017/07/24 15:16:26 - 建立入庫表結構.0 - ;
2017/07/24 15:16:26 - 建立入庫表結構.0 - 完成處理 (I=0, O=0, R=1, W=0, U=0, E=0)
2017/07/24 15:16:26 - 表數據抽取子做業 - 開始項[數據抽取]
2017/07/24 15:16:26 - 數據遷移 - 爲了轉換解除補丁開始  [數據遷移]
2017/07/24 15:16:26 - 表輸出.0 - Connected to database [testb] (commit=1000)
2017/07/24 15:16:26 - 表輸入.0 - Finished reading query, closing connection.
2017/07/24 15:16:26 - 表輸入.0 - 完成處理 (I=3, O=0, R=0, W=3, U=0, E=0)
2017/07/24 15:16:26 - 表輸出.0 - 完成處理 (I=0, O=3, R=3, W=3, U=0, E=0)
2017/07/24 15:16:26 - 表數據抽取子做業 - 完成做業項[數據抽取] (結果=[true])
2017/07/24 15:16:26 - 表數據抽取子做業 - 完成做業項[建立表結構] (結果=[true])
2017/07/24 15:16:26 - 表數據抽取子做業 - 完成做業項[代表成變量設置] (結果=[true])
2017/07/24 15:16:26 - 表數據抽取子做業 - 開始項[代表成變量設置]
2017/07/24 15:16:26 - 將記錄中的表名設置爲變量 - 爲了轉換解除補丁開始  [將記錄中的表名設置爲變量]
2017/07/24 15:16:26 - 從結果獲取記錄.0 - 完成處理 (I=0, O=0, R=1, W=1, U=0, E=0)
2017/07/24 15:16:26 - 設置變量.0 - Setting environment variables...
2017/07/24 15:16:26 - 設置變量.0 - Set variable TABLENAME to value [TESTN]
2017/07/24 15:16:26 - 設置變量.0 - Finished after 1 rows.
2017/07/24 15:16:26 - 設置變量.0 - 完成處理 (I=0, O=0, R=1, W=1, U=0, E=0)
2017/07/24 15:16:26 - 表數據抽取子做業 - 開始項[建立表結構]
2017/07/24 15:16:26 - 建立表結構 - 爲了轉換解除補丁開始  [建立表結構]
2017/07/24 15:16:26 - 表輸入.0 - Finished reading query, closing connection.
2017/07/24 15:16:26 - 表輸入.0 - 完成處理 (I=1, O=0, R=0, W=1, U=0, E=0)
2017/07/24 15:16:26 - 建立入庫表結構.0 - 數據庫鏈接名爲:testb
2017/07/24 15:16:26 - 建立入庫表結構.0 - 開始建立表：TESTN
2017/07/24 15:16:26 - 建立入庫表結構.0 - data.inputRowMeta=[ID String(10)], [NAME String(200)], [AGE BigNumber], [ADDRESS String(200)]
2017/07/24 15:16:26 - 建立入庫表結構.0 - sql=CREATE TABLE TESTN
2017/07/24 15:16:26 - 建立入庫表結構.0 - (
2017/07/24 15:16:26 - 建立入庫表結構.0 -   ID VARCHAR2(10)
2017/07/24 15:16:26 - 建立入庫表結構.0 - , NAME VARCHAR2(200)
2017/07/24 15:16:26 - 建立入庫表結構.0 - , AGE NUMBER
2017/07/24 15:16:26 - 建立入庫表結構.0 - , ADDRESS VARCHAR2(200)
2017/07/24 15:16:26 - 建立入庫表結構.0 - )
2017/07/24 15:16:26 - 建立入庫表結構.0 - ;
2017/07/24 15:16:26 - 建立入庫表結構.0 - CREATE TABLE TESTN
2017/07/24 15:16:26 - 建立入庫表結構.0 - (
2017/07/24 15:16:26 - 建立入庫表結構.0 -   ID VARCHAR2(10)
2017/07/24 15:16:26 - 建立入庫表結構.0 - , NAME VARCHAR2(200)
2017/07/24 15:16:26 - 建立入庫表結構.0 - , AGE NUMBER
2017/07/24 15:16:26 - 建立入庫表結構.0 - , ADDRESS VARCHAR2(200)
2017/07/24 15:16:26 - 建立入庫表結構.0 - )
2017/07/24 15:16:26 - 建立入庫表結構.0 - ;
2017/07/24 15:16:26 - 建立入庫表結構.0 - 完成處理 (I=0, O=0, R=1, W=0, U=0, E=0)
2017/07/24 15:16:26 - 表數據抽取子做業 - 開始項[數據抽取]
2017/07/24 15:16:26 - 數據遷移 - 爲了轉換解除補丁開始  [數據遷移]
2017/07/24 15:16:26 - 表輸出.0 - Connected to database [testb] (commit=1000)
2017/07/24 15:16:26 - 表輸入.0 - Finished reading query, closing connection.
2017/07/24 15:16:26 - 表輸入.0 - 完成處理 (I=3, O=0, R=0, W=3, U=0, E=0)
2017/07/24 15:16:27 - 表輸出.0 - 完成處理 (I=0, O=3, R=3, W=3, U=0, E=0)
2017/07/24 15:16:27 - 表數據抽取子做業 - 完成做業項[數據抽取] (結果=[true])
2017/07/24 15:16:27 - 表數據抽取子做業 - 完成做業項[建立表結構] (結果=[true])
2017/07/24 15:16:27 - 表數據抽取子做業 - 完成做業項[代表成變量設置] (結果=[true])
2017/07/24 15:16:27 - 數據庫遷移做業 - 開始項[成功]
2017/07/24 15:16:27 - 數據庫遷移做業 - 完成做業項[成功] (結果=[true])
2017/07/24 15:16:27 - 數據庫遷移做業 - 完成做業項[表數據抽取做業] (結果=[true])
2017/07/24 15:16:27 - 數據庫遷移做業 - 完成做業項[獲取表名流程] (結果=[true])
2017/07/24 15:16:27 - 數據庫遷移做業 - 任務執行完畢
2017/07/24 15:16:27 - Spoon - 任務已經結束.