KettleAPI-Java调用示例

1、Kettle API - Java 调用示例对向前兼容性的推荐 ：如果想要动态地创造 Transformation (例如 : 从元数据 )，使用XML文件方法(KTR)而不是使用API。 XML文件兼容Kettle所有版本，同样对job有效的。1. 下面的例子进行以下操作：1 创建 Transformation2 保存 Transformation 信息到 XML 文件3 获取在目标表操作的 SQL4 执行 Transformation5drop 目标表，使这个示例可以重复。/ 创建“复制表”的 transformation 元数据 .TransMeta transMeta = TransB

2、uilder.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("UTF

3、-8");dos.close();System.out.println("Saved transformation to file: "+fileName);/获取在目标表操作的SQLString sql = transMeta.getSQLStatementsString();/ 在目标表执行 sql:Database targetDatabase = new Database(transMeta.findDatabase(targetDatabaseName); targetDatabase.connect();targetDatabase.execState

4、ments(sql);/现在执行 transformationTrans trans = new Trans(transMeta);trans.execute(null); trans.waitUntilFinished();/为了重复测试 ,再次 drop 目标表 targetDatabase.execStatement("drop table "+targetTableName);targetDatabase.disconnect();面是创建 Transformation 的方法的代码：* Creates a new Transformation using inpu

5、t 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 要写入的列名 ( 要跟读取的列长度

6、相同 )* 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 targetTab

7、leName, 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(database

8、sXMLi); transMeta.addDatabase(databaseMeta);DatabaseMeta sourceDBInfo = transMeta.findDatabase(sourceDatabaseName); DatabaseMeta targetDBInfo = transMeta.findDatabase(targetDatabaseName);/ /添加注释 / String note = "Reads information from table " + sourceTableName+ " on database " +

9、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);/ 创建读取数据源的 step./String fromstepname = &

10、quot;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 += " &quo

11、t; selectSQL += sourceFieldsi + 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

12、);fromstep.setLocation(150, 100); fromstep.setDraw(true);fromstep.setDescription("Reads information from table " + sourceTableName + " on database " + sourceDBInfo + "");transMeta.addStep(fromstep);/ 添加 重命名 fields 的逻辑/ Use metadata logic in SelectValues, use SelectValue

13、Info./SelectValuesMeta svi = new SelectValuesMeta(); svi.allocate(0, 0, sourceFields.length);for (int i = 0; i < sourceFields.length; i+) svi.getMetaName()i = sourceFieldsi; svi.getMetaRename()i = targetFieldsi;String selstepname = "Rename field names"String selstepid = steploader.getSt

14、epPluginID(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); transM

15、eta.addTransHop(shi);fromstep = selstep;/ 创建 写数据的 step./ 添加 输出表 step./String tostepname = "write to " + targetTableName + "" TableOutputMeta toi = new TableOutputMeta();toi.setDatabase(targetDBInfo); toi.setTablename(targetTableName);toi.setCommitSize(200); toi.setTruncateTable(t

16、rue);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 " + t

17、argetDBInfo + "");transMeta.addStep(tostep);/ 添加连线 ./TransHopMeta hi = new TransHopMeta(fromstep, tostep); transMeta.addTransHop(hi);/ The transformation is complete, return it.return transMeta; catch (Exception e) newthrow new KettleException("An unexpected error occurred creating the transformation", e);2. 运行现有的 Transformation如果已经创建了 Transformation ，并且保存它在 .ktr 或 .xml 文件，能使用象下面的代码。 public static void runTransformation(String filename) try StepLoader.init();EnvUtil.environmentInit();TransMeta transMeta = new TransMeta(filename);Trans trans = n