JR 精品文章 - KETTLE?JAVA?API?实战记录

为什么要用Kettle和KETTLE JAVA API？
Kettle是什么？kettle：是一个开源ETL工具。kettle提供了基于java的图形化界面，使用很方便，kettle的ETL工具集合也比较多，常用的ETL工具都包含了。

为什么使用KETTLE JAVA API：就像kettle文档所说：KETTLE JAVA API : Program your own Kettle transformation，kettle提供了基于JAVA的脚步编写功能，可以灵活地自定义ETL过程，使自行定制、批量处理等成为可能，这才是一个程序员需要做的工作，而不仅是象使用word一样操作kettle用户界面。

KETTLE JAVA API 实战操作记录：

一、         搭建环境 :到http://www.kettle.be网站下载kettle的源码包，加压缩，例如解压缩到d:\kettle目录

二、         打开eclipse，新建一个项目，要使用jdk1.5.0，因为kettle的要使用System.getenv(),只有在jdk1.5.0才被支持。提起getenv（），好像有一段几起几落的记录，曾一度被抛弃，现在又被jdk1.5支持了.

三、         建一个class : TransBuilder.java,可以把d:\kettle\ extra\TransBuilder.java的内容原样拷贝到你的TransBuilder.java里。

四、         根据需要编辑源码。并需要对原程序进行如下修改，在头部增加：

import org.eclipse.swt.dnd.Transfer;

//这个包被遗漏了，原始位置kettle根目录\libswt\win32\swt.jar

//add by chq(www.chq.name) on  2006.07.20

（后来发现，不必加这个引用，因为编译时不需要）

五、         编译准备，在eclipse中增加jar包，主要包括（主要依据extra\TransBuilder.bat）：

\lib\kettle.jar
\libext\CacheDB.jar
\libext\SQLBaseJDBC.jar
\libext\activation.jar
\libext\db2jcc.jar
\libext\db2jcc_license_c.jar
\libext\edtftpj-1.4.5.jar
\libext\firebirdsql-full.jar
\libext\firebirdsql.jar
\libext\gis-shape.jar
\libext\hsqldb.jar
\libext\ifxjdbc.jar
\libext\javadbf.jar
\libext\jconn2.jar
\libext\js.jar
\libext\jt400.jar
\libext\jtds-1.1.jar
\libext\jxl.jar
\libext\ktable.jar
\libext\log4j-1.2.8.jar
\libext\mail.jar
\libext\mysql-connector-java-3.1.7-bin.jar
\libext\ojdbc14.jar
\libext\orai18n.jar
\libext\pg74.215.jdbc3.jar
\libext\edbc.jar

(注意 :下面这个包被遗漏了，要加上。原始位置kettle根目录\libswt\win32\swt.jar)
\libswt\win32\swt.jar 

六、         编译成功后，准备运行

为使程序不必登陆就可以运行，需要设置环境署文件：kettle.properties，位置在用户目录里，一般在 \Documents and Settings\用户\.kettle\,主要内容如下：

KETTLE_REPOSITORY=kettle@m80

KETTLE_USER=admin

KETTLE_PASSWORD=passwd

七、         好了，现在可以运行一下了，看看数据是不是已经拷贝到目标表了。

以下是运行时的控制台信息输出：



下面是自动生成的Transformation :



以下为修改后的程序源码：


--------------------------------------------------------------------------------

2. package name.chq.test;
4.  
6. import java.io.DataOutputStream;
8. import java.io.File;
10. import java.io.FileOutputStream;
12.  
14. import be.ibridge.kettle.core.Const;
16. import be.ibridge.kettle.core.LogWriter;
18. import be.ibridge.kettle.core.NotePadMeta;
20. import be.ibridge.kettle.core.database.Database;
22. import be.ibridge.kettle.core.database.DatabaseMeta;
24. import be.ibridge.kettle.core.exception.KettleException;
26. import be.ibridge.kettle.core.util.EnvUtil;
28. import be.ibridge.kettle.trans.StepLoader;
30. import be.ibridge.kettle.trans.Trans;
32. import be.ibridge.kettle.trans.TransHopMeta;
34. import be.ibridge.kettle.trans.TransMeta;
36. import be.ibridge.kettle.trans.step.StepMeta;
38. import be.ibridge.kettle.trans.step.StepMetaInterface;
40. import be.ibridge.kettle.trans.step.selectvalues.SelectValuesMeta;
42. import be.ibridge.kettle.trans.step.tableinput.TableInputMeta;
44. import be.ibridge.kettle.trans.step.tableoutput.TableOutputMeta;
46.  
48.  
50. //这个包被遗漏了，原始位置kettle根目录\libswt\win32\swt.jar
52. //add by chq([link=http://www.chq.name]www.chq.name[/link]) on  2006.07.20
54. //import org.eclipse.swt.dnd.Transfer; 
56.  
58. /**
60.  * Class created to demonstrate the creation of transformations on-the-fly.
62.  * 
64.  * @author Matt
66.  * 
68.  */
70. public class TransBuilder
72. {
74.     public static final String[] databasesXML = {
76.         "<?xml version=\"1.0\" encoding=\"UTF-8\"?>" +
78.         "<connection>" +
80.           "<name>target</name>" +
82.           "<server>192.168.17.35</server>" +
84.           "<type>ORACLE</type>" +
86.                      "<access>Native</access>" +
88.                      "<database>test1</database>" +
90.                      "<port>1521</port>" +
92.                      "<username>testuser</username>" +
94.                      "<password>pwd</password>" +
96.                      "<servername/>" +
98.                      "<data_tablespace/>" +
100.                      "<index_tablespace/>" +
102.                      "<attributes>" +
104.                        "<attribute><code>EXTRA_OPTION_MYSQL.defaultFetchSize</code><attribute>500</attribute></attribute>" +
106.                        "<attribute><code>EXTRA_OPTION_MYSQL.useCursorFetch</code><attribute>true</attribute></attribute>" +
108.                           "<attribute><code>PORT_NUMBER</code><attribute>1521</attribute></attribute>" +
110.                             "</attributes>" +
112.                        "</connection>" ,
114.          
116.         "<?xml version=\"1.0\" encoding=\"UTF-8\"?>" +
118.                          "<connection>" +
120.                                 "<name>source</name>" +
122.                                 "<server>192.168.16.12</server>" +
124.                                 "<type>ORACLE</type>" +
126.                                 "<access>Native</access>" +
128.                                 "<database>test2</database>" +
130.                                 "<port>1521</port>" +
132.                                 "<username>testuser</username>" +
134.                                 "<password>pwd2</password>" +
136.                                 "<servername/>" +
138.                                 "<data_tablespace/>" +
140.                                 "<index_tablespace/>" +
142.                                 "<attributes>" +
144.                                     "<attribute><code>EXTRA_OPTION_MYSQL.defaultFetchSize</code><attribute>500</attribute></attribute>" +
146.                                     "<attribute><code>EXTRA_OPTION_MYSQL.useCursorFetch</code><attribute>true</attribute></attribute>" +
148.                                        "<attribute><code>PORT_NUMBER</code><attribute>1521</attribute></attribute>" +
150.                                 "</attributes>" +
152.                          "</connection>" 
154.     };
156.  
158.     /**
160.      * Creates a new Transformation using input parameters such as the tablename to read from.
162.      * @param transformationName The name of the transformation
164.      * @param sourceDatabaseName The name of the database to read from
166.      * @param sourceTableName The name of the table to read from
168.      * @param sourceFields The field names we want to read from the source table
170.      * @param targetDatabaseName The name of the target database
172.      * @param targetTableName The name of the target table we want to write to
174.      * @param targetFields The names of the fields in the target table (same number of fields as sourceFields)
176.      * @return A new transformation
178.      * @throws KettleException In the rare case something goes wrong
180.      */
182.     public static final TransMeta buildCopyTable(
184.        String transformationName,String sourceDatabaseName, String sourceTableName, 
186.        String[] sourceFields, String targetDatabaseName, String targetTableName, 
188.        String[] targetFields)
190.       throws KettleException
192.     {
194.         LogWriter log = LogWriter.getInstance();
196.         EnvUtil.environmentInit();
198.         try
200.         {
202.             //
204.             // Create a new transformation...
206.             //
208.             TransMeta transMeta = new TransMeta();
210.             transMeta.setName(transformationName);
212.             
214.             // Add the database connections
216.  
218.             for (int i=0;i<databasesXML.length;i++)
220.             {
222.                 DatabaseMeta databaseMeta = new DatabaseMeta(databasesXML[i]);
224.                 transMeta.addDatabase(databaseMeta);
226.             }
228.             
230.             DatabaseMeta sourceDBInfo = transMeta.findDatabase(sourceDatabaseName);
232.             DatabaseMeta targetDBInfo = transMeta.findDatabase(targetDatabaseName);
234.  
236.             
238.             //
240.             // Add a note
242.             //
244.             String note = "Reads information from table [" + sourceTableName+ "] on database [" 
246.                             + sourceDBInfo + "]" + Const.CR;
248.             note += "After that, it writes the information to table [" + targetTableName + "] on database [" 
250.                             + targetDBInfo + "]";
252.             NotePadMeta ni = new NotePadMeta(note, 150, 10, -1, -1);
254.             transMeta.addNote(ni);
256.  
258.             // 
260.             // create the source step...
262.             //
264.             String fromstepname = "read from [" + sourceTableName + "]";
266.             TableInputMeta tii = new TableInputMeta();
268.             tii.setDatabaseMeta(sourceDBInfo);
270.             String selectSQL = "SELECT "+Const.CR;
272.             for (int i=0;i<sourceFields.length;i++)
274.             {
276.             /* modi by chq(www.chq.name): use * to replace the fields,经分析,以下语句可以处理‘*‘ */
278.                 if (i>0) 
280.                      selectSQL+=", "; 
282.                 else selectSQL+="  ";
284.                
286.                 selectSQL+=sourceFields[i]+Const.CR;
288.             }
290.             selectSQL+="FROM "+sourceTableName;
292.             tii.setSQL(selectSQL);
294.  
296.             StepLoader steploader = StepLoader.getInstance();
298.  
300.             String fromstepid = steploader.getStepPluginID(tii);
302.             StepMeta fromstep = new StepMeta(log, fromstepid, fromstepname, (StepMetaInterface) tii);
304.             fromstep.setLocation(150, 100);
306.             fromstep.setDraw(true);
308.             fromstep.setDescription("Reads information from table [" + sourceTableName 
310.                                      + "] on database [" + sourceDBInfo + "]");
312.             transMeta.addStep(fromstep);
314.  
316.             //
318.             // add logic to rename fields
320.             // Use metadata logic in SelectValues, use SelectValueInfo...
322.             //
324.             /* 不必改名或映射 add by chq(www.chq.name) on 2006.07.20
326.             SelectValuesMeta svi = new SelectValuesMeta();
328.             svi.allocate(0, 0, sourceFields.length);
330.             for (int i = 0; i < sourceFields.length; i++)
332.             {
334.                 svi.getMetaName()[i] = sourceFields[i];
336.                 svi.getMetaRename()[i] = targetFields[i];
338.             }
340.  
342.             String selstepname = "Rename field names";
344.             String selstepid = steploader.getStepPluginID(svi);
346.             StepMeta selstep = new StepMeta(log, selstepid, selstepname, (StepMetaInterface) svi);
348.             selstep.setLocation(350, 100);
350.             selstep.setDraw(true);
352.             selstep.setDescription("Rename field names");
354.             transMeta.addStep(selstep);
356.  
358.             TransHopMeta shi = new TransHopMeta(fromstep, selstep);
360.             transMeta.addTransHop(shi);
362.             fromstep = selstep; //设定了新的起点 by chq([link=http://www.chq.name]www.chq.name[/link]) on 2006.07.20
364.             */
366.             // 
368.             // Create the target step...
370.             //
372.             //
374.             // Add the TableOutputMeta step...
376.             //
378.             String tostepname = "write to [" + targetTableName + "]";
380.             TableOutputMeta toi = new TableOutputMeta();
382.             toi.setDatabase(targetDBInfo);
384.             toi.setTablename(targetTableName);
386.             toi.setCommitSize(200);
388.             toi.setTruncateTable(true);
390.  
392.             String tostepid = steploader.getStepPluginID(toi);
394.             StepMeta tostep = new StepMeta(log, tostepid, tostepname, (StepMetaInterface) toi);
396.             tostep.setLocation(550, 100);
398.             tostep.setDraw(true);
400.             tostep.setDescription("Write information to table [" + targetTableName + "] on database [" + targetDBInfo + "]");
402.             transMeta.addStep(tostep);
404.  
406.             //
408.             // Add a hop between the two steps...
410.             //
412.             TransHopMeta hi = new TransHopMeta(fromstep, tostep);
414.             transMeta.addTransHop(hi);
416.  
418.             // OK, if we're still here: overwrite the current transformation...
420.             return transMeta;
422.         }
424.         catch (Exception e)
426.         {
428.             throw new KettleException("An unexpected error occurred creating the new transformation", e);
430.         }
432.     }
434.  
436.     /**
438.      * 1) create a new transformation
440.      * 2) save the transformation as XML file
442.      * 3) generate the SQL for the target table
444.      * 4) Execute the transformation
446.      * 5) drop the target table to make this program repeatable
448.      * 
450.      * @param args
452.      */
454.     public static void main(String[] args) throws Exception
456.     {
458.        EnvUtil.environmentInit();
460.         // Init the logging...
462.         LogWriter log = LogWriter.getInstance("TransBuilder.log", true, LogWriter.LOG_LEVEL_DETAILED);
464.         
466.         // Load the Kettle steps & plugins 
468.         StepLoader stloader = StepLoader.getInstance();
470.         if (!stloader.read())
472.         {
474.             log.logError("TransBuilder",  "Error loading Kettle steps & plugins... stopping now!");
476.             return;
478.         }
480.         
482.         // The parameters we want, optionally this can be 
484.         String fileName = "NewTrans.xml";
486.         String transformationName = "Test Transformation";
488.         String sourceDatabaseName = "source";
490.         String sourceTableName = "testuser.source_table";
492.         String sourceFields[] = { 
494.                "*" 
496.                };
498.  
500.         String targetDatabaseName = "target";
502.         String targetTableName = "testuser.target_table";
504.         String targetFields[] = { 
506.                "*"
508.                };
510.  
512.         
514.         // Generate the transformation.
516.         TransMeta transMeta = TransBuilder.buildCopyTable(
518.                 transformationName,
520.                 sourceDatabaseName,
522.                 sourceTableName,
524.                 sourceFields,
526.                 targetDatabaseName,
528.                 targetTableName,
530.                 targetFields
532.                 );
534.         
536.         // Save it as a file:
538.         String xml = transMeta.getXML();
540.         DataOutputStream dos = new DataOutputStream(new FileOutputStream(new File(fileName)));
542.         dos.write(xml.getBytes("UTF-8"));
544.         dos.close();
546.         System.out.println("Saved transformation to file: "+fileName);
548.  
550.         // OK, What's the SQL we need to execute to generate the target table?
552.         String sql = transMeta.getSQLStatementsString();
554.         
556.         // Execute the SQL on the target table:
558.         Database targetDatabase = new Database(transMeta.findDatabase(targetDatabaseName));
560.         targetDatabase.connect();
562.         targetDatabase.execStatements(sql);
564.         
566.         // Now execute the transformation...
568.         Trans trans = new Trans(log, transMeta);
570.         trans.execute(null);
572.         trans.waitUntilFinished();
574.         
576.         // For testing/repeatability, we drop the target table again
578.         /* modi by chq([link=http://www.chq.name]www.chq.name[/link]) on  2006.07.20 不必删表
580.         //targetDatabase.execStatement("drop table "+targetTableName);
582.         targetDatabase.disconnect();
584.     }
586.  
588.  
590. }
592.  

 

相关文档：

 

[翻译]KETTLE JAVA API :编程定制自己的Kettle转换（transformation）