Java应用程序中集成Kettle
摘要：本文主要讨论如何在你自己的Java应用程序中集成Kettle
如果你需要在自己的Java应用程序中集成Kettle，一般来说有两种应用需求，一种是通过纯设计器来设计ETL转换任务，然后保存成某种格式，比如xml或者在数据库中都可以，然后自己调用程序解析这个格式，执行这种转换，是比较抽象的一种执行方式，ETL里面转换了什么东西我们并不关心，只关心
它有没有正常执行。另一种是通过完全编程的方式来实现，详细的控制每一个
步骤，需要知道转换执行的成功与否，这种方式可能需要更多的理解kettle的API以便更好的跟你的应用程序紧密结合，不过难度也比较大，可以很好的定
制你的应用程序，代价自然是入门门槛比较高。本文主要向你解释第一种Kettle的集成方式，文中所列出的代码节选自pentaho，不过应用程序本身跟pentaho没有什么关系。
Pentaho集成kettle的代码主要是两个类，KettleSystemListener和KettleComponent，看名字就猜出KettleSystemListener主要是起的作用，
它主要负责初始化kettle的一些环境变量，这个类主要包含四个方法:startup()，readProperties()，environ
mentInit()，shutdown()，程序入口自然是startup()方法，然后它会调用environmentInit()方法，这个方法就调用readProperties()方法读一
个配置文件kettle。properties，这个文件主要记录者kettle运行时可以调用的一
些环境变量，关于kettle。properties文件怎么用，第二篇文章“使用Kettle设计
动态转换”有提到，readProperties()方法读完这个文件之后就把里面的键值对转
换成变量传给kettle运行环境。当kettle运行完了之后就调用shutdown()方法结
束转换。KettleSystemListener相对逻辑比较简单，就不多介绍，下面主要介绍
重点类：
KettleComponent
KettleComponent的方法主要有三种类型，一类是用来初始化工作，做一些
验证工作，第二类是执行转换的方法，也是主要需要讨论的方法，第三类是取
得数据结果的，有时候你需要得到转换的结果交给下一个步骤处理。下面分别
讨论这三类方法。
初始化
KettleComponent的初始化工作主要是验证这个转换，包括有validateSystemSettings()，init()，validateAction()，全部都是public方法，validateSystemSettings()会检查kettle使用何种方式来连接资源库。
kettle有两种方式连接资源库，一种是纯数据库式，也就是你所有的转换全部都保存在一个数据库中，一般你在开始使用kettle的时候，它都会要求你建
立一个资源仓库，这个资源仓库的连接方式就是你的数据库连接，你需要能够有相应的数据库驱动和对应的连接用户名和密码。另外一种连接方式是使用文本文件，也就是xml文件，在做完任何转换之后，我们都可以把转换或者Job
变成xml文件输出，这个输出文件包含你所有转换的全部信息。
在示例应用中使用的是文件的连接方式，下面看一下初始化的一段代码:
Boolean useRepository=
files
admin
admin
可以看到其中的l.file上面的一段注释，如果这个值为空会默认使用$HOME/.l文件当作资源库的连接文件，由于示例中使用的是文本文件所以没有用数据库连接，下面的repository.userid和repository.password是指的kettle的资源库连接的用户名和密码，一般默认安装就两个，admin/admin和guest/guest，这里的用户名和密码不是连接数据库的用户名和密码，连接数据库的用户名和密码是在另外一个文件l。file指定的值所定义的
一般默认的kettle安装并且运行了一段时间之后，会在$HOME/.kettle目录下创建一些文件，如果你要在自己的系统中集成kettle的话，也需要保留这些文件，当然不一定位置是在原来的位置，关键是要让kettle知道这些文件放在哪。
执行转换
当读完了这些配置文件并且验证了之后，KettleComponent就开始把前面读到的转换文件或者资源库类型变成Kettle的API，这主要是在executeAction（）方法里面进行，它当然根据连接方式也分两种执行类型：
1.文本执行方式
2.资源库连接方式
文本执行方式需要接受一个你指定的运行转换的文件或者Job的文件，然后把这个xml文件解析成Kettle能够执行的模式，
根据执行的类型又可以分成两种：
1.Trans任务
java创建文件2.Job任务
两个执行的逻辑差不多，下面先介绍Trans的执行方式：
执行Trans任务
transMeta=new TransMeta(fileAddress,repository,true);
transMeta.setFilename(fileAddress);
然后它会调用:
executeTransformation(TransMeta transMeta,LogWriter logWriter)
这个方法是真正的把前面的来的transMeta转换成trans对象，等待下一步的执行：
Trans trans=new Trans(logWriter,transMeta);
List Steps();
for(int stepNo=0;stepNo<stepList.size();stepNo++){
StepMetaDataCombi step=((stepNo);
if(step.stepname.equals(stepName)){
①Row StepFields(stepName);
//create the metadata that the Pentaho result set needs
String fieldNames[]=FieldNames();
String columns[][]=new String[1][fieldnames.length];
for(int column=0;column<fieldnames.length;column++){
columns[0][column]=fieldNames[column];
}
②MemoryMetaData metaData=new MemoryMetaData(columns,null);
results=new MemoryResultSet(metaData);
//add ourself as a row listener
③step.step.addRowListener(this);
foundStep=true;
break;
}
}
1.Row对象是kettle用来表示一行数据的标准对象，跟jdbc取出来的一条数据转化后成为的一个POJO是一样的。里面可以包含多个字段。
2.MemoryMetaData对象是pentaho特有的，是专门用来返回ETL任务执行后的结果的，与标准的JDBC里面的resultSet对应的resultSetMetaData是一样的。
3.对于如何处理数据的一个Listener，实现的是一个RowListener，数据是每一行每一行处理的，后面会介绍如果需要输出数据怎么取得这些输出数据。如果不需要放回任何对象，则从1处开始都可以不要，只要初始化step对象即可。
所有的step对象都已经初始化之后就可以开始执行了，
trans.startThreads();
trans.waitUntilFinished();
结束之后还有一些清理工作就不列出了。
执行Job任务
执行Job任务之前还是会读取Job任务的描述文件，然后把这个描述文件（kettle的.ktr文件）变成一个xml文档的dom：
org.w3c.dom.Document DomFromString(jobXmlStr);
之后也是初始化对应的元数据对象JobMeta
jobMeta=new JobMeta(FirstChild(),repository);
得到了jobMeta之后就可以执行这个Job了，这里跟trans是一样的。
job=new Job(Instance(),repository,jobMeta);
由于Job一般都没有什么返回值，所以Job不需要初始化它下面的对象，直接开始运行就可以了
job.start();
job.waitUntilFinished(5000000);
连接资源库
连接资源库使用的是connectToRepository()方法，先取得RepositoriesMeta 对象，然后根据你在setting。xml文件里面定义的repository的名字来连接对应的repository。理论上来说我们一般都只使用一个repository，但如果在产品中需要使用多个repository的话，你需要自己配置多个repository的名字和对应的用户名和密码。只列出几行关键代码，
repositoriesMeta=new RepositoriesMeta(logWriter);
repositoryMeta=repositoriesMeta.findRepository(repositoryName);
repository=new Repository(logWriter,repositoryMeta,userInfo);
userInfo=new UserInfo(repository,username,password);
从资源库读取Trans
连接到资源库之后自然是想办法读取数据库的表，把里面的记录转换成为Trans对象，使用的是loadTransformFromRepository，这个方法的函数原型需要解释一下：
TransMetaloadTransformFromRepository(String directoryName,String transformationName,Repository repository,LogWriter logWriter)
第一个参数String directoryName代表是你储存转换的目录，当你使用kettle图形界面的时候，点击repository菜单的explorer repository，你会发现你所有的东西都是存储在一个虚拟的类似与目录结构的地方，其中包括database connections，transformations，job，users等，所以你需要的是指定你连接的目录位置，你也可以在目录里面再创建目录。
String transformationName自然指的就是你转换的名字。
Repository repository指的是你连接的资源库。
LogWriter logWriter指定你的日志输出，这个log指的是你kettle转换的日志输出，不是应用程序本身的输出。
读取TransMeta的步骤也相对比较简单
DirectoryTree()。
findDirectory(directoryName);
transMeta=new TransMeta(repository,transformationName, repositoryDirectory);
从资源库读取Job
从资源库读取Job跟Trans的步骤基本是一样的，同样需要指定你存储Job 的目录位置。
JobMeta loadJobFromRepository(String directoryName,String jobName,
Repository repository,LogWriter logWriter)
读取结果集

版权声明：本站内容均来自互联网，仅供演示用，请勿用于商业和其他非法用途。如果侵犯了您的权益请与我们联系QQ:729038198，我们将在24小时内删除。