君，已阅读到文档的结尾了呢~~
Kettle进阶开发(胡春阳)开发,帮助,胡春阳,沈春阳,霍春阳,奚春阳,张春阳,石春阳
扫扫二维码，随身浏览文档
手机或平板扫扫即可继续访问
Kettle进阶开发(胡春阳)
举报该文档为侵权文档。
举报该文档含有违规或不良信息。
反馈该文档无法正常浏览。
举报该文档为重复文档。
推荐理由：
将文档分享至：
分享完整地址
文档地址：
粘贴到BBS或博客
flash地址：
支持嵌入FLASH地址的网站使用
html代码：
&embed src='/DocinViewer--144.swf' width='100%' height='600' type=application/x-shockwave-flash ALLOWFULLSCREEN='true' ALLOWSCRIPTACCESS='always'&&/embed&
450px*300px480px*400px650px*490px
支持嵌入HTML代码的网站使用
您的内容已经提交成功
您所提交的内容需要审核后才能发布，请您等待！
3秒自动关闭窗口本帖子已过去太久远了，不再提供回复功能。后使用快捷导航没有帐号？
查看: 1506|回复: 0
Kettle二次开发例程 transBuilder.java
论坛徽章:32
本帖最后由 wangpugood 于
23:30 编辑
Kettle二次开发例程 transBuilder.
第一步：创建一个新的transMeta对象，该对象用于描述整个数据抽取过程。
第二步：如果数据抽取过程涉及到数据库操作，则定义并设置好相关数据库，并作为属性添加到transMeta对象中
第三步：建立数据来源step，将数据来源step对象作为属性添加到TransMeta对象中。
第四步：如果需要对所有数据做处理，则建立处理过程step，将该step作为属性添加到TransMeta对象中。
第五步：建立一个数据流hop对象，该数据流对象源自第三步，流至第四步，用于step之间的数据传输。该数据流hop对象也需要作为属性添加到TransMeta对象中。
第六步：建立目标数据step，该对象描述了最终数据的去向。将该对象最为属性添加到TransMeta对象中
第七步：与第五步相似，在处理过程step与目标数据step之间建立数据流hop对象。
第八步：执行整个TransMeta对象，完成数据抽取过程
基本工作流程
具体代码：
import java.io.DataOutputS
import java.io.F
import java.io.FileOutputS
importbe.ibridge.kettle.core.C
importbe.ibridge.kettle.core.LogW
importbe.ibridge.kettle.core.NotePadM
importbe.ibridge.kettle.core.database.D
importbe.ibridge.kettle.core.database.DatabaseM
importbe.ibridge.kettle.core.exception.KettleE
importbe.ibridge.kettle.core.util.EnvU
importbe.ibridge.kettle.trans.StepL
importbe.ibridge.kettle.trans.T
importbe.ibridge.kettle.trans.TransHopM
importbe.ibridge.kettle.trans.TransM
importbe.ibridge.kettle.trans.step.StepM
importbe.ibridge.kettle.trans.step.StepMetaI
importbe.ibridge.kettle.trans.step.selectvalues.SelectValuesM
importbe.ibridge.kettle.trans.step.tableinput.TableInputM
importbe.ibridge.kettle.trans.step.tableoutput.TableOutputM
public class TransBuilder
//这是对数据库的定义
&&public static final String[]databasesXML = {
//目标数据库的定义
&&?xml version=\&1.0\&encoding=\&UTF-8\&?&& +
& && & &&connection&& +
& && && &&&&&name&target&/name&&+
& && && &&&&&server&localhost&/server&&+
& && && &&&&&type&MS&/type&&+
& && && &&&&&access&Native&/access&&+
& && && &&&&&database&test&/database&&+
& && && &&&&&port&1433&/port&&+
& && && &&&&&username&matt&/username&&+
& && && &&&&&password&abcd&/password&&+
& && && &&&/connection&&,
//来数据库定义
& && && &&&?xml version=\&1.0\&encoding=\&UTF-8\&?&& +
& && && &&&connection&& +
& && && && & &&name&source&/name&&+
& && && && & &&server&localhost&/server&&+
& && && && & &&type&MYSQL&/type&&+
& && && && & &&access&Native&/access&&+
& && && && & &&database&test&/database&&+
& && && && & &&port&3306&/port&&+
& && && && & &&username&matt&/username&&+
& && && && & &&password&abcd&/password&&+
& && && &&&&&/connection&&
Public static final TransMeta buildCopyTable(String transformationName,String sourceDatabaseName, String sourceTableName, String[]sourceFields, String targetDatabaseName, String targetTableName,String[] targetFields) throws KettleException
//对环境变量进行初始化
& && & EnvUtil.environmentInit();
&&// Create a new transformation...
&&//第一步：创建一个transMeta对象
TransMeta transMeta = new TransMeta();
&&transMeta.setName(transformationName);
// Add the database connections
//第二步：创建数据库对象，将数据库对象作为属性添加到transMeta对象中
for (int i=0;i&databasesXML.i++)
& && && &&&{
& && && && && &DatabaseMeta databaseMeta = newDatabaseMeta(databasesXML);
& && && && && &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...
//第三步： 创建一个源数据step，该setp是一个来自数据库的源
String fromstepname = &read from [& + sourceTableName + &]&;
& && && &&&TableInputMeta tii = newTableInputMeta();& && && &//一个表输入源
//该表所在的数据库
& && && &&&tii.setDatabaseMeta(sourceDBInfo);& && && && && && &&&
& && && &&&String selectSQL = &SELECT &+Const.CR;
& && && &&&for (int i=0;i&sourceFields.i++)
& && && &&&{
& && && && && &if (i&0) selectSQL+=&, &; else selectSQL+=& &;
& && && && && &selectSQL+=sourceFields+Const.CR;
& && && &&&}
& && && &&&selectSQL+=&FROM &+sourceTableN
//设定取数据的sql语句
& && && &&&tii.setSQL(selectSQL);& && && && && && && && && && && && && &
// 以下将源数据step作为属性添加到transMeta对象中
& && && &&&StepLoader steploader = StepLoader.getInstance();
& && && &&&String fromstepid = steploader.getStepPluginID(tii);
& && && &&&StepMeta fromstep = new StepMeta(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...
//第四步：创建一个数据处理对象，此处为一个select对象。该对象用于选择指定的源数据//的列，将其中的数据送到目标数据库的指定列
& && &SelectValuesMeta svi = new SelectValuesMeta();
& && && &&&svi.allocate(0, 0, sourceFields.length);
& && && &&&for (int i = 0; i & sourceFields. i++)
& && && &&&{
& && && && && &svi.getMetaName() =sourceFields;& && && && && && && & //设定源列
& && && && && &svi.getMetaRename() =targetFields;& && && && && && & //设定目标列
& && && &&&}
& && && &&&//以下将该select对象作为属性添加到transMeta对象中
& && && &&&String selstepname = &Rename field names&;
& && && &&&String selstepid = steploader.getStepPluginID(svi);
& && && &&&StepMeta selstep = new StepMeta(selstepid, selstepname,(StepMetaInterface) svi);
& && && &&&selstep.setLocation(350, 100);
& && && &&&selstep.setDraw(true);
& && && &&&selstep.setDescription(&Rename field names&);
& && && &&&transMeta.addStep(selstep);
& && && &&&//第五步：创建一个数据流对象，用于连接源和select对象并将该数据流对象作为属//性添加到transMeta对象中
& && && &&&TransHopMeta shi = new TransHopMeta(fromstep, selstep);
& && && &&&transMeta.addTransHop(shi);
& && && &&&fromstep =
// 第六步：Create the target step...
// Add the TableOutputMeta step...
& && && &&&//创建一个目标对象，此处目标为数据库输出
& && && &&&String tostepname = &write to [& + targetTableName + &]&;
& && && &&&TableOutputMeta toi = newTableOutputMeta();& && && && && && &//创建一个表输出对象
& && && &&&toi.setDatabaseMeta(targetDBInfo);& && && && && && && && && && && && & //设置该表所在数据库
& && && &&&toi.setTablename(targetTableName);& && && && && && && && && && && && &//设置目标表的表名。若目标数据库中不存在该表，则创建表
& && && &&&toi.setCommitSize(200);
& && && &&&toi.setTruncateTable(true);
& && && & //将该目标对象作为属性添加到transMeta对象中
& && && &&&String tostepid = steploader.getStepPluginID(toi);
& && && &&&StepMeta tostep = new StepMeta(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...
// 第七步：建立一个数据流对象，用于连接select和目标
& && && &&&TransHopMeta hi = new TransHopMeta(fromstep, tostep);
& && && &&&transMeta.addTransHop(hi);
& && && &&&// OK, if we're still here: overwrite thecurrent transformation...
& && && &&&return transM
& &&&catch (Exception e)
& && && &&&throw new KettleException(&An unexpected error occurred creatingthe new transformation&, e);
//第八步：整体执行
&&publicstatic 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!&);
& && && &&&
& &// The parameters we want, optionally thiscan be
& && & String fileName = &NewTrans.xml&;
& && & String transformationName = &Test Transformation&;
& && & String sourceDatabaseName = &source&;
& && & String sourceTableName = &Customer&;
& && & String sourceFields[] = {
& && && && && &&customernr&,
& && && && && &&Name&,
& && && && && &&firstname&,
& && && && && &&lang&,
& && && && && &&sex&,
& && && && && &&street&,
& && && && && &&housnr&,
& && && && && &&bus&,
& && && && && &&zipcode&,
& && && && && &&location&,
& && && && && &&country&,
& && && && && &&date_of_birth&
& && && &&&};
& && & String targetDatabaseName = &target&;
& && & String targetTableName = &Cust&;
& && & String targetFields[] = {
& && && && && &&CustNo&,
& && && && && &&LastName&,
& && && && && &&FirstName&,
& && && && && &&Lang&,
& && && && && &&gender&,
& && && && && &&Street&,
& && && && && &&Housno&,
& && && && && &&busno&,
& && && && && &&ZipCode&,
& && && && && &&City&,
& && && && && &&Country&,
& && && && && &&BirthDate&
& && && &&&};
& && & // Generate thetransformation. 产生转换
& && & TransMeta transMeta = TransBuilder.buildCopyTable(
& && && && && &transformationName,
& && && && && &sourceDatabaseName,
& && && && && &sourceTableName,
& && && && && &sourceFields,
& && && && && &targetDatabaseName,
& && && && && &targetTableName,
& && && && && &targetFields
& && && && && &);
& && & // Save it as afile:
& && & String xml = transMeta.getXML();
& && & DataOutputStream dos = new DataOutputStream(newFileOutputStream(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 togenerate the targettable?
& && & String sql = transMeta.getSQLStatementsString();
& && & // Execute the SQL on the targettable:
& && & Database targetDatabase = newDatabase(transMeta.findDatabase(targetDatabaseName));
& && & targetDatabase.connect();
& && & targetDatabase.execStatements(sql);& && && && && && &
//执行transMeta对象，进行数据抽取工作& && &&&
& && & // Now execute thetransformation...
& && & Trans trans = new Trans(log, transMeta);
& && & trans.execute(null);
& && & trans.waitUntilFinished();
& && & // For testing/repeatability, we drop thetarget table again
& && & targetDatabase.execStatement(&drop table &+targetTableName);
& && & targetDatabase.disconnect();温馨提示！由于新浪微博认证机制调整，您的新浪微博帐号绑定已过期，请重新绑定！&&|&&
向自己定好的目标前进！！！
LOFTER精选
网易考拉推荐
用微信&&“扫一扫”
将文章分享到朋友圈。
用易信&&“扫一扫”
将文章分享到朋友圈。
阅读(991)|
用微信&&“扫一扫”
将文章分享到朋友圈。
用易信&&“扫一扫”
将文章分享到朋友圈。
历史上的今天
在LOFTER的更多文章
loftPermalink:'',
id:'fks_',
blogTitle:'数据对接—kettle使用之八',
blogAbstract:'& & & &之前由于一些事情耽误了，博客一直没有更新，今天开始继续完善kettle的学习，这一节开始了解转换控件的使用，首先从值映射控件开始。值映射：',
blogTag:'',
blogUrl:'blog/static/',
isPublished:1,
istop:false,
modifyTime:2,
publishTime:5,
permalink:'blog/static/',
commentCount:0,
mainCommentCount:0,
recommendCount:0,
bsrk:-100,
publisherId:0,
recomBlogHome:false,
currentRecomBlog:false,
attachmentsFileIds:[],
groupInfo:{},
friendstatus:'none',
followstatus:'unFollow',
pubSucc:'',
visitorProvince:'',
visitorCity:'',
visitorNewUser:false,
postAddInfo:{},
mset:'000',
remindgoodnightblog:false,
isBlackVisitor:false,
isShowYodaoAd:false,
hostIntro:'向自己定好的目标前进！！！',
hmcon:'1',
selfRecomBlogCount:'0',
lofter_single:''
{list a as x}
{if x.moveFrom=='wap'}
{elseif x.moveFrom=='iphone'}
{elseif x.moveFrom=='android'}
{elseif x.moveFrom=='mobile'}
${a.selfIntro|escape}{if great260}${suplement}{/if}
{list a as x}
推荐过这篇日志的人：
{list a as x}
{if !!b&&b.length>0}
他们还推荐了：
{list b as y}
转载记录：
{list d as x}
{list a as x}
{list a as x}
{list a as x}
{list a as x}
{if x_index>4}{break}{/if}
${fn2(x.publishTime,'yyyy-MM-dd HH:mm:ss')}
{list a as x}
{if !!(blogDetail.preBlogPermalink)}
{if !!(blogDetail.nextBlogPermalink)}
{list a as x}
{if defined('newslist')&&newslist.length>0}
{list newslist as x}
{if x_index>7}{break}{/if}
{list a as x}
{var first_option =}
{list x.voteDetailList as voteToOption}
{if voteToOption==1}
{if first_option==false},{/if}&&“${b[voteToOption_index]}”&&
{if (x.role!="-1") },“我是${c[x.role]}”&&{/if}
&&&&&&&&${fn1(x.voteTime)}
{if x.userName==''}{/if}
网易公司版权所有&&
{list x.l as y}
{if defined('wl')}
{list wl as x}{/list}}