KETTLE实现数据的删除和更新

一、实现目标

　　源数据库的数据更新或者删除之后，目标数据库的数据跟着更新或删除，整体流程截图如下：

　　 KETTLE实现数据的删除和更新

一、准备工作

源数据库ORACLE 目标数据库MongoDB，在源数据库添加删除、更新触发器

二、操作步骤

添加表输入组件，连接ORACLE触发器记录表
添加JAVA代码组件，进行步骤跳转，根据输入的数据判断是删除或者更新，如果是删除，则跳转至MongoDB Delete步骤中，如果是更新的话，跳转至字段选择步骤中。JAVA代码中的详细信息如下：

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

}