Kettle——转换步骤信息统计两种实现

Kettle-步骤数据处理数量记录获取

      Kettle获取数据处理条数记录日志信息：以前的途径是配置转换的日志信息来保存到数据库中，这种方式不是很灵活。

      有没有一种类似步骤方式来进行获取呢？Kettle在5.0版本中提供了一个步骤：在【统计】-【转换步骤信息统计】。

       【转换步骤信息统计】步骤使用方式也很简单，如下图所示：

       【转换步骤信息统计】点击获取步骤，选择进行步骤信息统计的步骤，若是没有步骤，可以手写指定。右侧图为输出统计信息字段配置（字段名称）

       运行结果：

但是这个步骤无法获取前面步骤的字段，因此进行了修改，使其支持获取前面步骤的相关字段。相关代码参见：https://git.oschina.net/zokapper/Kettle-plugins

另外一种方式：利用【java代码】步骤实现效果，这个和上面的插件原理大致一样。这种方式更灵活（随时改代码），但是缺点也很明显，不如插件方式简单直观，使用起来复杂。如下图所示

import org.pentaho.di.trans.*;
import org.pentaho.di.trans.step.*;
import java.util.*;
import org.pentaho.di.i18n.BaseMessages;

private StepInterface previousStep;
private Object[] previousRow;
private Object[] r;
private Date startDate;
private boolean previousStepStatus;

public boolean processRow(StepMetaInterface smi, StepDataInterface sdi) throws KettleException
{
    Object[] row = getRow();
    if (first)
    {
        init();
        if(row!=null){
            startDate = new Date();
            StepStatus stepStatus = new StepStatus(previousStep);
            r = createOutputRow(row, data.outputRowMeta.size());
            int index = getInputRowMeta().size();
            //父step读取数量
            r[index++]=Long.valueOf(stepStatus.getLinesRead());
            //父step输入数量
            r[index++]=Long.valueOf(stepStatus.getLinesWritten());
            //父step入库数量
            r[index++]=Long.valueOf(stepStatus.getLinesOutput());
            //父step更新数量
            r[index++]=Long.valueOf(stepStatus.getLinesUpdated());
            //父step拒绝数量
            r[index++]=Long.valueOf(stepStatus.getLinesRejected());

            //开始时间
            r[index++] = startDate;
            //结束时间
            //r[index++] = startDate;
            putRow(data.outputRowMeta, r);
        }
        first = false;
    }
    if (row==null) {
        if (previousRow != null) {// no more input to be expected.
            StepStatus stepStatus = new StepStatus(previousStep);
            r = createOutputRow(previousRow, data.outputRowMeta.size());
            int index = getInputRowMeta().size();
            //父step读取数量
            r[index++]=Long.valueOf(stepStatus.getLinesRead());
            //父step输入数量
            r[index++]=Long.valueOf(stepStatus.getLinesWritten());
            //父step入库数量
            r[index++]=Long.valueOf(stepStatus.getLinesOutput());
            //父step更新数量
            r[index++]=Long.valueOf(stepStatus.getLinesUpdated());
            //父step拒绝数量
            r[index++]=Long.valueOf(stepStatus.getLinesRejected());
            //开始时间
            r[index++] = startDate;
            //结束时间
            r[index++] = new Date();
            putRow(data.outputRowMeta, r);
    }
        setOutputDone();
        return false;
    }
    previousRow = row;  
    return true;
}
public void init(){
    //监控步骤的名称
    String previousStepName="ANA_OUT_DAILY";
    Trans trans = getTrans();
    for (int j = 0; j