KETTLE实现数据的删除和更新

核桃露子 發表於 2015-11-3 17:08:00

一、实现目标
　　源数据库的数据更新或者删除之后，目标数据库的数据跟着更新或删除，整体流程截图如下：
　　<img src="https://images2015.cnblogs.com/blog/467150/201511/467150-20151103165025039-1676779811.png" alt="">
一、准备工作
源数据库ORACLE  目标数据库MongoDB，在源数据库添加删除、更新触发器
二、操作步骤
<ol>
<li>添加表输入组件，连接ORACLE触发器记录表</li>
<li>添加JAVA代码组件，进行步骤跳转，根据输入的数据判断是删除或者更新，如果是删除，则跳转至MongoDB Delete步骤中，如果是更新的话，跳转至字段选择步骤中。JAVA代码中的详细信息如下：</li>
<li>
<div class="cnblogs_code">
<pre>import java.util.List;

import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.row.RowDataUtil;
import org.pentaho.di.core.row.RowMeta;
import org.pentaho.di.core.row.RowMetaInterface;
import org.pentaho.di.core.row.ValueMeta;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransMeta;

private Object[] previousRow;//上一行
private RowSet t1 = null;//业务表步骤
private RowSet t2 = null;//删除步骤

public boolean processRow(StepMetaInterface smi, StepDataInterface sdi) throws KettleException
{
Object[] r = getRow(); //获取输入行

if ( first ) {
 if ( getInputRowMeta() == null ) {
 setOutputDone();//设置输出完成
 return false;
 }
}

if ( r == null ) { // 如果当前行为null
 if ( previousRow != null ) {//如果上一行不为null
 //是最后一行

 boolean valid=true;
 previousRow = createOutputRow(previousRow, data.outputRowMeta.size());
 Trans trans=getTrans();//获取转换实例
 if (trans != null){
 String sync_val = get(Fields.In, "ID").getString(previousRow);//获取ID
 trans.setVariable("LAST_SYNC_VAL", sync_val);//设置变量的值
 }
 String OpType = get(Fields.In, "DATATYPE").getString(previousRow);//获取操作类型是删除还是更新
 String keyid= get(Fields.In, "DATAID").getString(previousRow);//获取操作类型是删除还是更新

 //Object[] rowData = RowDataUtil.allocateRowData(data.outputRowMeta.size());
 //get(Fields.Out, "KEYID").setValue(rowData,keyid);
 //putRowTo(data.outputRowMeta, previousRow,t2);

 if(OpType.equals("UPDATE")){//验证通过 
 putRowTo(data.outputRowMeta, previousRow,t1);
 }
 else
 {
 putRowTo(data.outputRowMeta, previousRow,t2);
 }

 }
 setOutputDone();//设置输出完成
 return false;//返回false表示不用再继续处理processRow
 }

if ( !first ) {//不是第一次执行，因为第一次执行时previousRow一定是Null
 //不是最后一行
 boolean valid=true;
 String OpType = get(Fields.In, "DATATYPE").getString(previousRow);//获取操作类型是删除还是更新
 String keyid= get(Fields.In, "DATAID").getString(previousRow);//获取操作类型是删除还是更新

 //Object[] rowData = RowDataUtil.allocateRowData(data.outputRowMeta.size());
 //get(Fields.Out, "KEYID").setValue(rowData,keyid);
 //putRowTo(data.outputRowMeta, previousRow,t2);
 if(OpType.equals("UPDATE")){
 putRowTo(data.outputRowMeta, previousRow,t1);
 }
 else
 {
 putRowTo(data.outputRowMeta, previousRow,t2);
 }
}
previousRow = r;//把当前行设为下一次执行的上一行
if ( first ) {//如果是首次执行
 first = false;
 t1 = findTargetRowSet("dataupdate");//业务表步骤
 t2 = findTargetRowSet("datadelete");//数据删除步骤
 }

return true;//返回true表示还要继续处理processRow
}</pre>
</div>
3.如果跳转至了MongoDB Delete,则根据ID对目标库进行删除。Mongodb delete组件配置如下：<img style="line-height: 1.5" src="https://images2015.cnblogs.com/blog/467150/201511/467150-20151103165314133-422581866.png" alt="">
</li>
</ol>
　　JSON query中的{ID:"?{DATAID}"}表示删除ID等于传进来的参数DATAID的所有数据，Execute for each row要选择上，表示执行每一行数据。          
　　4.如果通过JAVA代码2判断为更新的话，则流程将跳转至字段选择组件，只获取主键ID，此步骤非常重要，因为要根据ID去源表中获取等更新的那条数据。
<img src="https://images2015.cnblogs.com/blog/467150/201511/467150-20151103165901992-1000623393.png" alt="">
5.选择表输入组件，该步骤是根据上一步传入的ID获取待更新的那一条数据
<img src="https://images2015.cnblogs.com/blog/467150/201511/467150-20151103170039102-34982897.png" alt="">
PS:获取SQL查询语句：此处写入SQL语句，里边的?是变量替换，下边要勾选上"替换SQL语句里的变量"，从步骤插入数据要选择上一步，勾选上执行每一行。
6.下边的步骤：流查询、JAVA代码是对数据进行清洗，字典替换，此处不再解释
7.最后一步：Mongodb output输出需要详细设置
<img src="https://images2015.cnblogs.com/blog/467150/201511/467150-20151103170606774-1921240685.png" alt="">
output options选项卡勾选update  modifier update
<img src="https://images2015.cnblogs.com/blog/467150/201511/467150-20151103170611867-186038240.png" alt="">
Mongo文档字段配置：ID为主键匹配字段，匹配字段更新为Y 修改器设置为N/A表示不对主键更新
  
来源：https://www.cnblogs.com/nyzhai/p/4933773.html

頁: [1]

琼殿技术论坛's Archiver

KETTLE实现数据的删除和更新