java 调用 kettle api 实现增量同步
内容导读
互联网集市收集整理的这篇技术教程文章主要介绍了java 调用 kettle api 实现增量同步，小编现在分享给大家，供广大互联网技能从业者学习和参考。文章包含7589字，纯文字阅读大概需要11分钟。
内容图文
java 调用kettle api 实现数据库数据增量同步 和一般方式不同 这里是使用 java代码绘制job 与 transtion
项目背景：
数据库A处于外网 里面有test表 数据库B处于内网 里面test 如果数据库A的test表数据方式 增加 删除 修改 操作 触发器 会记录下来对应的数据并保存在 对应的临时表中 kettle获取临时表里面的数据 把数据同步到数据库B的test表中。
同步流程
流程很简单 就是数据发生变化 同步变化的数据就可以了
设计流程
- 创建表的触发器 增加 删除 修改
- 使用kettle 生成数据文件
- 摆渡文件到B库所在的服务
- kettle 读取数据文件 实现数据同步
核心代码
- A库 这边kettle 操作 因为代码很多 此处只展示核心代码
//设置转化的名称
transMeta.setName("dbToExcel");
//添加转换的数据库连接
DatabaseMeta databaseMeta = new DatabaseMeta(databasesXML);
transMeta.addDatabase(databaseMeta);
TableInputMeta tableInputMeta = new TableInputMeta();
tableInputMeta.setDatabaseMeta(databaseMeta);
tableInputMeta.setSQL(sqls.get(i));
StepMeta tableInputMetaStep = new StepMeta("表输入", tableInputMeta);
tableInputMetaStep.setDraw(true);
tableInputMetaStep.setLocation(100, 100);
transMeta.addStep(tableInputMetaStep);
SelectValuesMeta selectValuesMeta = new SelectValuesMeta();
selectValuesMeta.allocate(selectFields.length, 0, 0);
selectValuesMeta.setSelectFields(selectFields);
StepMeta selectValuesMetaStep = new StepMeta("字段选择",selectValuesMeta);
selectValuesMetaStep.setDraw(true);
selectValuesMetaStep.setLocation(300, 100);
transMeta.addStep(selectValuesMetaStep);
transMeta.addTransHop(new TransHopMeta(tableInputMetaStep, selectValuesMetaStep));
RowsToResultMeta rowsToResultMeta = new RowsToResultMeta();
StepMeta rowsToResultMetaStep = new StepMeta("复制记录到结果",rowsToResultMeta);
rowsToResultMetaStep.setDraw(true);
rowsToResultMetaStep.setLocation(500, 100);
transMeta.addStep(rowsToResultMetaStep);
transMeta.addTransHop(new TransHopMeta(selectValuesMetaStep,rowsToResultMetaStep));
UserDefinedJavaClassMeta userDefinedJavaClassMeta = new UserDefinedJavaClassMeta();
List<UserDefinedJavaClassDef> definitions = new ArrayList<UserDefinedJavaClassDef>();
String source= "import java.util.List;\n" +
"import org.pentaho.di.trans.steps.userdefinedjavaclass.UsageParameter;\n" +
"import com.wondersoft.dt.dbtrans.formatter.FormatterFactory;\n" +
"import com.wondersoft.dt.dbtrans.formatter.FormatterInterface;\n" +
"\n" +
"public boolean processRow(StepMetaInterface smi, StepDataInterface sdi) throws KettleException{\n" +
"\n" +
" Object[] r = getRow();\n" +
"\t\t\t\t if (r == null) { \n" +
"\t\t\t\t\t\t\t\t setOutputDone();\n" +
"\t\t\t\t\t\t\t\t return false; \n" +
"\t\t\t\t\t\t } \n" +
"\t\t\t\t\t\t Object[] outputRow = createOutputRow(r, data.outputRowMeta.size());\n" +
" UserDefinedJavaClassMeta rmeta = meta ;\n" +
" List usageParameters = rmeta.getUsageParameters();\n" +
" for(int i=0;i<usageParameters.size();i++){\n" +
" UsageParameter usageParameter = (UsageParameter)usageParameters.get(i);\n" +
"\t\t \t String tag = usageParameter.tag;\n" +
"\t\t \t String value = usageParameter.value;\n" +
" String targetColumnType = usageParameter.description;\n" +
" Object v = get(Fields.In, tag).getObject(r); \n" +
" FormatterInterface formatterInterface = FormatterFactory.getFormatter(value);\n" +
" if(formatterInterface != null){\n" +
" get(Fields.Out, tag).setValue(outputRow, formatterInterface.format(v,targetColumnType)); \n" +
" get(Fields.In, tag).getValueMeta().setType(ValueMetaInterface.TYPE_STRING);\n" +
" continue;\n" +
" }\n" +
"\t\t\t\t\t \t get(Fields.Out, tag).setValue(outputRow, v); \n" +
"\t\t }\n" +
"\t\t \n" +
"\t\t\t\t\t\t\t putRow(data.outputRowMeta, outputRow); \n" +
"\t\t\t\t\t\t\t return true; \n" +
"\n" +
"}";
UserDefinedJavaClassDef classDef = new UserDefinedJavaClassDef(UserDefinedJavaClassDef.ClassType.TRANSFORM_CLASS,"Processor",source);
definitions.add(classDef);
userDefinedJavaClassMeta.replaceDefinitions(definitions);
userDefinedJavaClassMeta.setUsageParameters(usageParameters);
StepMeta userDefinedJavaClassMetaStep = new StepMeta("行处理",userDefinedJavaClassMeta);
userDefinedJavaClassMetaStep.setDraw(true);
userDefinedJavaClassMetaStep.setLocation(700, 100);
transMeta.addStep(userDefinedJavaClassMetaStep);
transMeta.addTransHop(new TransHopMeta(rowsToResultMetaStep,userDefinedJavaClassMetaStep));
ExcelOutputMeta excelOutputMeta = new ExcelOutputMeta();
excelOutputMeta.setDefault();
excelOutputMeta.setAutoSizeColums(true);
excelOutputMeta.setFileName(outputFilePathi);
StepMeta excelOutputMetaStep = new StepMeta("导出至Excel", excelOutputMeta);
excelOutputMetaStep.setDraw(true);
excelOutputMetaStep.setLocation(900, 100);
transMeta.addStep(excelOutputMetaStep);
transMeta.addTransHop(new TransHopMeta(userDefinedJavaClassMetaStep, excelOutputMetaStep));
Trans trans = new Trans(transMeta);
//执行转换
trans.execute(null);
trans.waitUntilFinished();
insert or update 逻辑操作
insertUpdateMeta.setDatabaseMeta(databaseMeta);
// 设置操作的表
insertUpdateMeta.setTableName(tableName);
//如果不存在主键则进行全字段匹配
if(primaryColumnNames.length == 0) {
primaryColumnNames = columnNames;
}
// 设置用来查询的关键字
insertUpdateMeta.setKeyLookup(primaryColumnNames);
insertUpdateMeta.setKeyStream(primaryColumnNames);
String[] keyStream2 = new String[columnNames.length];
String[] keyCondition = new String[columnNames.length];
for(int m = 0 ;m<primaryColumnNames.length;m++) {
keyStream2[m] = "";
keyCondition[m] = "=";
}
insertUpdateMeta.setKeyStream2(keyStream2);// 一定要加上
insertUpdateMeta.setKeyCondition(keyCondition);
Boolean[] updateOrNot = new Boolean[columnNames.length];
for(int m = 0 ;m<columnNames.length;m++) {
updateOrNot[m] = true;
}
insertUpdateMeta.setUpdateLookup(columnNames);
insertUpdateMeta.setUpdateStream(columnNames);
insertUpdateMeta.setUpdate(updateOrNot);
StepMeta toTableStep = new StepMeta("导出至数据库表", insertUpdateMeta);
toTableStep.setLocation(550, 100);
toTableStep.setDraw(true);
transMeta.addStep(toTableStep);
transMeta.addTransHop(new TransHopMeta(stepMeta, toTableStep));
delete 操作
DeleteMeta deleteMeta = new DeleteMeta();
deleteMeta.setDatabaseMeta(databaseMeta);
// 设置操作的表
deleteMeta.setTableName(tableName);
//如果不存在主键则进行全字段匹配
if(primaryColumnNames.length == 0) {
primaryColumnNames = columnNames;
}
deleteMeta.setKeyLookup(primaryColumnNames);
deleteMeta.setKeyStream(primaryColumnNames);
String[] keyStream2 = new String[columnNames.length];
String[] keyCondition = new String[columnNames.length];
for(int m = 0 ;m<primaryColumnNames.length;m++) {
keyStream2[m] = "";
keyCondition[m] = "=";
}
deleteMeta.setKeyStream2(keyStream2);// 一定要加上
deleteMeta.setKeyCondition(keyCondition);
deleteMeta.setChanged(true);
StepMeta toTableStep = new StepMeta("导出至数据库表", deleteMeta);
toTableStep.setLocation(550, 100);
toTableStep.setDraw(true);
transMeta.addStep(toTableStep);
transMeta.addTransHop(new TransHopMeta(stepMeta, toTableStep));
注意点
没有什么复杂的操作 主要是注意 放获取数据文件的时间 区分有主键和无主键的情况 insert update 和delete 使用的类是不同的。
主要的事情说三遍
这里是代码绘制 不是调用okr文件
这里是代码绘制 不是调用okr文件
这里是代码绘制 不是调用okr文件
内容总结
以上是互联网集市为您收集整理的java 调用 kettle api 实现增量同步全部内容，希望文章能够帮你解决java 调用 kettle api 实现增量同步所遇到的程序开发问题。 如果觉得互联网集市技术教程内容还不错，欢迎将互联网集市网站推荐给程序员好友。
内容备注
版权声明：本文内容由互联网用户自发贡献，该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容， 请发送邮件至 gblab@vip.qq.com 举报，一经查实，本站将立刻删除。
内容手机端
扫描二维码推送至手机访问。
来源：【匿名】