Kettle模版文件简单使用示例

KETTLE版本：4.3
1.本示例效果。
  a.可以对两个库中结果相同的表进行数据迁移。
  b.可选择性把源表的字段抽取到目标表。
  c.一个模版即可完成一系列的数据抽取。
解决方案：
  定义库表迁移流程，即，《表输入->表输出》可下载附件查看。
模版所需参数：
1.源表数据连接名称（模版文件中已经保存的连接），
2.源表抽取sql，
3.目标表连接。
4.目标表名称（事先创建好）。
5.抽取sql中要入库的字段数组。
6.目标表对应源表的字段数组。

示例具体代码：

package org.pentaho.di.trans;

import java.util.List;

import org.pentaho.di.core.KettleEnvironment;
import org.pentaho.di.core.database.DatabaseMeta;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.logging.Log4jFileAppender;
import org.pentaho.di.core.logging.LogWriter;
import org.pentaho.di.trans.step.StepMeta;
import org.pentaho.di.trans.steps.tableinput.TableInputMeta;
import org.pentaho.di.trans.steps.tableoutput.TableOutputMeta;

public class TestTransMain {
	
	public static void tableToTableSimpleTemplateDemo(
			String tableInputConneName, String sql, String tableOutConnneName,
			String dest_tablename,String [] streamFields,String [] tableOutFields) throws Exception
	{
		String filename = "D:\\etl_tem\\4.3测试流程\\表输入表输出模版.ktr";
		
		String arguments[] = {};
		
		try {

			TransMeta meta = new TransMeta(filename);
			
			DatabaseMeta tableInputDbMeta = null;
			DatabaseMeta tableoutDbMeta = null;
			
			List<DatabaseMeta> list = meta.getDatabases();
			
			for(DatabaseMeta dbMeta : list)
			{
				if(dbMeta.getName().equals(tableInputConneName))
				{
					tableInputDbMeta = dbMeta;
				}
				
				if(dbMeta.getName().equals(tableOutConnneName))
				{
					tableoutDbMeta = dbMeta;
				}
				
				if(tableoutDbMeta!=null && tableInputDbMeta!=null) break;
			}

			if(tableoutDbMeta==null || tableInputDbMeta==null)
			{
				System.out.println("指定的数据源名称在模版文件中不存在!");
				
				return ;
			}
			
			StepMeta stepMeta = meta.findStep("表输入");
			
			TableInputMeta tableInput_meta = (TableInputMeta)stepMeta.getStepMetaInterface();
			tableInput_meta.setDatabaseMeta(tableInputDbMeta);
			tableInput_meta.setSQL(sql);
			
			stepMeta = meta.findStep("表输出");
			TableOutputMeta tableout_meta = (TableOutputMeta)stepMeta.getStepMetaInterface();
			tableout_meta.setDatabaseMeta(tableoutDbMeta);
			tableout_meta.setTablename(dest_tablename);
			tableout_meta.setFieldStream(streamFields);
			tableout_meta.setFieldDatabase(tableOutFields);
			
			Trans trans = new Trans(meta);
			
			trans.prepareExecution(arguments);

			trans.addTransListener(new TransAdapter() {
				public void transFinished(Trans trans) throws KettleException {
					System.out.println("流程执行完成");
				}
			});
			
			trans.startThreads();
			
			trans.waitUntilFinished();
			
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	public static void main(String[] args) {
		String tableInputConneName = "153ora";
		String sql = "SELECT  ID, NAME, SEX FROM PERSON"; 
		String tableOutConnneName = "153ora";
		String dest_tablename = "TTT_Person";
		String [] streamFields = {"ID","NAME","SEX"};
		String [] tableOutFields = {"ID","NAME","SEX"};
		
		try {
			KettleEnvironment.init();

			Log4jFileAppender fileAppender = LogWriter.createFileAppender(
					"D:\\etl_tem\\4.3测试流程\\test.log", true);
			LogWriter.getInstance().addAppender(fileAppender);
			
			tableToTableSimpleTemplateDemo(tableInputConneName, sql, tableOutConnneName, dest_tablename, streamFields, tableOutFields);
			
			fileAppender.close();
		} catch (Exception e) {
			e.printStackTrace();
		}
		
		System.out.println("主函数结束");
	}
}