Kettle API- Java调用示例

Kettle API - Java调用示例

对向前兼容性的推荐 ：如果想要动态地创造Transformation (例如:从元数据)，使用XML文件方法(KTR)而不是使用API。 XML文件兼容Kettle所有版本，同样对job有效的。

1. 下面的例子进行以下操作：

1创建Transformation

2保存Transformation信息到XML文件 3获取在目标表操作的SQL 4执行Transformation

5drop目标表，使这个示例可以重复。

// 创建“复制表”的 transformation元数据.

TransMeta transMeta = TransBuilder.buildCopyTable( transformationName, sourceDatabaseName, sourceTableName, sourceFields,

targetDatabaseName, targetTableName, targetFields );

// 保存为文件:

String xml = transMeta.getXML();

DataOutputStream dos = new DataOutputStream(new FileOutputStream(new File(fileName))); dos.write(xml.getBytes(\dos.close();

System.out.println(\

//获取在目标表操作的SQL

String sql = transMeta.getSQLStatementsString();

// 在目标表执行sql:

Database targetDatabase = new Database(transMeta.findDatabase(targetDatabaseName)); targetDatabase.connect();

targetDatabase.execStatements(sql); //现在执行 transformation

Trans trans = new Trans(transMeta); trans.execute(null);

trans.waitUntilFinished();

//为了重复测试,再次drop目标表

targetDatabase.execStatement(\targetDatabase.disconnect();

下面是创建Transformation的方法的代码： /**

* Creates a new Transformation using input parameters such as the tablename to read from. * @param transformationName transformation的名称 * @param sourceDatabaseName 输入的 database 名称 * @param sourceTableName 要读取的表名 * @param sourceFields 要读取的列名

* @param targetDatabaseName 目标database名 * @param targetTableName要写入的表名

* @param targetFields要写入的列名(要跟读取的列长度相同) * @return A new transformation metadata object

* @throws KettleException In the rare case something goes wrong */

public static final TransMeta buildCopyTable(String transformationName, String sourceDatabaseName, String sourceTableName, String[] sourceFields, String targetDatabaseName, String targetTableName, String[] targetFields) throws KettleException { EnvUtil.environmentInit(); try { // Create a new transformation... // TransMeta transMeta = new TransMeta(); transMeta.setName(transformationName); // 添加数据库连接 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); // //添加注释 // String note = \sourceDBInfo + \ note += \that, it writes the information to table [\targetTableName + \on

database [\ NotePadMeta ni = new NotePadMeta(note, 150, 10, -1, -1); transMeta.addNote(ni); // // 创建读取数据源的 step... // String fromstepname = \ TableInputMeta tii = new TableInputMeta(); tii.setDatabaseMeta(sourceDBInfo); String selectSQL = \ for (int i = 0; i < sourceFields.length; i++) { if (i > 0) selectSQL += \ selectSQL += sourceFields[i] + Const.CR; } selectSQL += \ 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(\information from table [\+ sourceTableName + \database [\ transMeta.addStep(fromstep); // // 添加 重命名 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 = \

String selstepid = steploader.getStepPluginID(svi); StepMeta selstep = new StepMeta(log, selstepid, selstepname, (StepMetaInterface) svi); selstep.setLocation(350, 100); selstep.setDraw(true); selstep.setDescription(\ transMeta.addStep(selstep); TransHopMeta shi = new TransHopMeta(fromstep, selstep); transMeta.addTransHop(shi); fromstep = selstep; // // 创建 写数据的 step... // // // 添加 输出表 step... // String tostepname = \ 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(\[\ transMeta.addStep(tostep); // // 添加连线... // TransHopMeta hi = new TransHopMeta(fromstep, tostep); transMeta.addTransHop(hi); // The transformation is complete, return it... return transMeta;

} catch (Exception e) { throw new KettleException(%unexpected error occurred creating the new transformation\ } }

2. 运行现有的Transformation

如果已经创建了Transformation，并且保存它在.ktr或.xml文件，能使用象下面的代码。

public static void runTransformation(String filename) { try {

StepLoader.init();

EnvUtil.environmentInit();

TransMeta transMeta = new TransMeta(filename); Trans trans = new Trans(transMeta);

trans.execute(null); // You can pass arguments instead of null. trans.waitUntilFinished(); if ( trans.getErrors() > 0 ) {

throw new RuntimeException( \ } }

catch ( KettleException e ) { System.out.println(e); } }

本文来源：https://www.bwwdw.com/article/arjv.html