KETTLE JAVA API学习

/**KETTLE是一个开源的ETL工具，运行本程序，应该确保在SQL Server中有相应数据库asop2,asop3，在asop2中有表department_user,但在表 asop3中有没有表test2均行。（相应的数据库名，表名，用户名，密码可看程序修改）。另外，应该在spoon.bat中打开后建立存储库与asop2及asop3两个连接（连接名应与数据库名一致）。感觉KETTLE没有转化不同数据类型的数据，这一点上应该没有Beeload好吧。**/

package test;

import java.io.DataOutputStream;
import java.io.File;
import java.io.FileOutputStream;
import be.ibridge.kettle.core.Const;
import be.ibridge.kettle.core.LogWriter;
import be.ibridge.kettle.core.NotePadMeta;
import be.ibridge.kettle.core.database.Database;
import be.ibridge.kettle.core.database.DatabaseMeta;
import be.ibridge.kettle.core.exception.KettleException;
import be.ibridge.kettle.core.util.EnvUtil;
import be.ibridge.kettle.trans.StepLoader;
import be.ibridge.kettle.trans.Trans;
import be.ibridge.kettle.trans.TransHopMeta;
import be.ibridge.kettle.trans.TransMeta;
import be.ibridge.kettle.trans.step.StepMeta;
import be.ibridge.kettle.trans.step.StepMetaInterface;
import be.ibridge.kettle.trans.step.selectvalues.SelectValuesMeta;
import be.ibridge.kettle.trans.step.tableinput.TableInputMeta;
import be.ibridge.kettle.trans.step.tableoutput.TableOutputMeta;
import org.eclipse.swt.dnd.Transfer;

public class TransBuilder {

//用于SQL Server的测试

public static final String[] databasesXML = {
"<?xml version=/"1.0/" encoding=/"UTF-8/"?>" + "<connection>"
+ "<name>asop2</name>" + "<server>localhost</server>"
+ "<type>MSSQL</type>" + "<access>Native</access>"
+ "<database>asop2</database>" + "<port>1433</port>"
+ "<username>sa</username>"
+ "<password>password1234</password>" + "</connection>",

"<?xml version=/"1.0/" encoding=/"UTF-8/"?>" + "<connection>"
+ "<name>asop3</name>" + "<server>localhost</server>"
+ "<type>MSSQL</type>" + "<access>Native</access>"
+ "<database>asop3</database>" + "<port>1433</port>"
+ "<username>sa</username>"
+ "<password>password1234</password>" + "</connection>" };


/** //用于Oracle的测试
public static final String[] databasesXML = {
"<?xml version=/"1.0/" encoding=/"UTF-8/"?>" + "<connection>"
+ "<name>asop2</name>" + "<server>localhost</server>"
+ "<type>MSSQL</type>" + "<access>Native</access>"
+ "<database>asop2</database>" + "<port>1433</port>"
+ "<username>sa</username>"
+ "<password>password1234</password>" + "</connection>",

"<?xml version=/"1.0/" encoding=/"UTF-8/"?>" + "<connection>"
+ "<name>oracledb</name>" + "<server>localhost</server>"
+ "<type>oracle</type>" + "<access>Native</access>"
+ "<database>oracledb</database>" + "<port>1521</port>"
+ "<username>system</username>"
+ "<password>password</password>" + "</connection>" };
**/

public static final TransMeta buildCopyTable(String transformationName,
String sourceDatabaseName, String sourceTableName,
String[] sourceFields, String targetDatabaseName,
String targetTableName, String[] targetFields)
throws KettleException {
LogWriter log = LogWriter.getInstance();
EnvUtil.environmentInit();
try {
// Create a new transformation...
TransMeta transMeta = new TransMeta();
transMeta.setName(transformationName);
// Add the database connections
for (int i = 0; i < databasesXML.length; i++) {
DatabaseMeta databaseMeta = new DatabaseMeta(databasesXML[i]);
transMeta.addDatabase(databaseMeta);
}
DatabaseMeta sourceDBInfo = transMeta
.findDatabase(sourceDatabaseName);
DatabaseMeta targetDBInfo = transMeta
.findDatabase(targetDatabaseName);
// Add a note
String note = "Reads information from table [" + sourceTableName
+ "] on database [" + sourceDBInfo + "]" + Const.CR;
note += "After that, it writes the information to table ["
+ targetTableName + "] on database [" + targetDBInfo + "]";
NotePadMeta ni = new NotePadMeta(note, 150, 10, -1, -1);
transMeta.addNote(ni);
// create the source step...
String fromstepname = "read from [" + sourceTableName + "]";
TableInputMeta tii = new TableInputMeta();
tii.setDatabaseMeta(sourceDBInfo);
String selectSQL = "SELECT " + Const.CR;
for (int i = 0; i < sourceFields.length; i++) {
if (i > 0)
selectSQL += ", ";
else
selectSQL += " ";
selectSQL += sourceFields[i] + Const.CR;
}
selectSQL += "FROM " + sourceTableName;
tii.setSQL(selectSQL);
StepLoader steploader = StepLoader.getInstance();
String fromstepid = steploader.getStepPluginID(tii);
StepMeta fromstep = new StepMeta(log, fromstepid, fromstepname,
(StepMetaInterface) tii);
fromstep.setLocation(150, 100);
fromstep.setDraw(true);
fromstep.setDescription("Reads information from table ["
+ sourceTableName + "] on database [" + sourceDBInfo + "]");
transMeta.addStep(fromstep);
// add logic to rename fields
// Use metadata logic in SelectValues, use SelectValueInfo...
SelectValuesMeta svi = new SelectValuesMeta();
svi.allocate(0, 0, sourceFields.length);
//改名，那如何改类型啊
for (int i = 0; i < sourceFields.length; i++) {
svi.getMetaName()[i] = sourceFields[i];
svi.getMetaRename()[i] = targetFields[i];

}
String selstepname = "Rename field names";
String selstepid = steploader.getStepPluginID(svi);
StepMeta selstep = new StepMeta(log, selstepid, selstepname,
(StepMetaInterface) svi);
selstep.setLocation(350, 100);
selstep.setDraw(true);
selstep.setDescription("Rename field names");
transMeta.addStep(selstep);
TransHopMeta shi = new TransHopMeta(fromstep, selstep);
transMeta.addTransHop(shi);
fromstep = selstep;
// Add the TableOutputMeta step...
String tostepname = "write to [" + targetTableName + "]";
TableOutputMeta toi = new TableOutputMeta();
toi.setDatabase(targetDBInfo);
toi.setTablename(targetTableName);
toi.setCommitSize(200);
toi.setTruncateTable(true);
String tostepid = steploader.getStepPluginID(toi);
StepMeta tostep = new StepMeta(log, tostepid, tostepname,
(StepMetaInterface) toi);
tostep.setLocation(550, 100);
tostep.setDraw(true);
tostep.setDescription("Write information to table ["
+ targetTableName + "] on database [" + targetDBInfo + "]");
transMeta.addStep(tostep);
// Add a hop between the two steps...
TransHopMeta hi = new TransHopMeta(fromstep, tostep);
transMeta.addTransHop(hi);
// OK, if we're still here: overwrite the current transformation...
return transMeta;
} catch (Exception e) {
throw new KettleException(
"An unexpected error occurred creating the new transformation",
e);
}
}

public static void main(String[] args) throws Exception {
EnvUtil.environmentInit();
// Init the logging...
LogWriter log = LogWriter.getInstance("TransBuilder.log", true,
LogWriter.LOG_LEVEL_DETAILED);

// Load the Kettle steps & plugins
StepLoader stloader = StepLoader.getInstance();
if (!stloader.read()) {
log.logError("TransBuilder",
"Error loading Kettle steps & plugins... stopping now!");
return;
}

// The parameters we want, optionally this can be
String fileName = "NewTrans.xml";
String transformationName = "Test Transformation";
String sourceDatabaseName = "asop2";
String sourceTableName = "department_user";
String sourceFields[] = { "department_id", "login_name", "title",
"name" };

String targetDatabaseName = "asop3"; //SQL Server中
//String targetDatabaseName = "oracledb"; //orcale中
String targetTableName = "test2";

String targetFields[] = { "department", "username", "title", "fullname2" };
// Generate the transformation.
TransMeta transMeta = TransBuilder.buildCopyTable(transformationName,
sourceDatabaseName, sourceTableName, sourceFields,
targetDatabaseName, targetTableName, targetFields);
// Save it as a file:
String xml = transMeta.getXML();
DataOutputStream dos = new DataOutputStream(new FileOutputStream(
new File(fileName)));
dos.write(xml.getBytes("UTF-8"));
dos.close();
System.out.println("Saved transformation to file: " + fileName);
// OK, What's the SQL we need to execute to generate the target table?
String sql = transMeta.getSQLStatementsString();
// Execute the SQL on the target table:
DatabaseMeta dbMeta = transMeta.findDatabase(targetDatabaseName);
Database targetDatabase = new Database(dbMeta);
targetDatabase.connect();
targetDatabase.execStatements(sql);
System.out.println("执行的SQL语句是："+sql);
// Now execute the transformation...
Trans trans = new Trans(log, transMeta);
trans.execute(null);
trans.waitUntilFinished();
// For testing/repeatability, we drop the target table again
//targetDatabase.execStatement("drop table " + targetTableName);
targetDatabase.disconnect();
}

}