先把环境搭建好&＃xff0c;这里可以参考我之前的博客&＃xff0c;或者比较好的参考文献&＃xff1a;

https://blog.csdn.net/mar_ljh/article/details/89333215

kettle转换步骤插件至少需要实现四个接口&＃xff1a;

org.pentaho.di.trans.step.StepInterface 负责数据处理&＃xff0c;转换和流转。这里面主要由processRow()方法来处理。

org.pentaho.di.trans.step.StepDataInterface 数据处理设计的具体数据&＃xff0c;以及对数据的状态的设置和回收

org.pentaho.di.trans.step.StepMetaInterface 元数据的处理&＃xff0c;加载xml&＃xff08;负责将Dialog数据加载到xml&＃xff09;&＃xff0c;校验&＃xff0c;主要是对一个步骤的定义的基本数据。

org.pentaho.di.trans.step.StepDialogInterface 提供GUI/dialog编译步骤的元数据。

StepInterface的实现类以插件的功能先关命名&＃xff1a;*.java

StepDataInterface的实现类&＃xff1a;*Data.java

StepMetaInterface实现类&＃xff1a;*Meta.java

StepDialogInterface实现类&＃xff1a;*Dialog.java

Demon测试Template stepTemplate stepTemplate stepTemplate step

这里对Meta后缀的类重点讲解&＃xff08;弄清楚这个类&＃xff0c;其实已经解决了trans插件开发的都大部分问题&＃xff09;&＃xff1a;

注意&＃xff1a;getXML是保存、loadXML是读取&＃xff0c;这2个方法基本是属于模板类的代码&＃xff0c;跟着其他项目一样写即可。本质使用xml来定义data数据格式

如果需要对插件进行错误处理步骤&＃xff0c;实现错误分流&＃xff08;实现&＃xff1a;主步骤输出、错误处理步骤&＃xff09;&＃xff0c;需要重写父类方法&＃xff0c;返回true即可&＃xff1a;

&＃64;Overridepublic boolean supportsErrorHandling() {return true;}

这里可以参考一个kettle已有插件的源码&＃xff1a;&＃xff08;数据检验&＃xff09;

Dialog类&＃xff1a;/kettle-7.1-eclipse/ui/org/pentaho/di/ui/trans/steps/validator/ValidatorDialog.java

其他类&＃xff1a;/kettle-7.1-eclipse/engine/org/pentaho/di/trans/steps/validator/包下有&＃xff1a;

KettleValidatorException.java

Validation.java

Validator.java

ValidatorData.java

ValidatorMeta.java

messages/

国际化相关&＃xff1a;messages/:

messages_en_US.properties

messages_fr_FR.properties

messages_it_IT.properties

messages_ja_JP.properties

messages_ko_KR.properties

messages_zh_CN.properties

开始执行&＃xff0c;调用&＃xff1a;public boolean processRow( StepMetaInterface smi, StepDataInterface sdi ) throws KettleException

里面使用Object[]r &＃61; getRow(); // get row, set busy!获取一行记录

返回值是boolean&＃xff0c;其中有个first来标记是否为第一行。

返回值&＃xff1a;

有个RunThread类内run方法通过while来不断循环调用processRow方法。

在processRow中使用&＃xff1a;

关于日志&＃xff1a;BaseStep有如下方法&＃xff0c;直接使用即可打印在kettle控制台:

logMinimal、logBasic、logDetailed、logDebug等

并在代码中判断级别&＃xff1a;

if (log.isDetailed())

...

if (log.isDebug())

...

if (log.isRowlevel())

...

具体meta类中一些变量&＃xff0c;可以调试查看&＃xff1a;在*&＃xff08;按照kettlt命名规范&＃xff09;类中&＃xff1a;

meta &＃61; (MetaDataPluginMeta) smi;

data &＃61; (MetaDataPluginData) sdi;

1&＃xff0c;meta拿到了dialog中填写的值

2&＃xff0c;调用getRow()方法&＃xff0c;直接拿到对应的一行数据&＃xff0c;这里input是表输入&＃xff0c;这里直接拿到了数据库中第一行数据&＃xff08;数据&＃xff0c;不是表头&＃xff09;

3&＃xff0c;其中first是BaseStep中定义&＃xff0c;当第一行时&＃xff0c;first标记为true

4&＃xff0c;获取表头信息调用&＃xff1a; List metaList &＃61; getInputRowMeta().getValueMetaList();

这里上面声明了一个data对象是用户存储数据的&＃xff0c;data里面有字段来保存表头信息&＃xff1a;data.outputRowMeta &＃61; (RowMetaInterface) getInputRowMeta().clone();

5.获取组件名称&＃xff1a;getStepname()

6.getMetaStore()

7:关于&＃xff1a;getFields方法&＃xff1a;

public void getFields( RowMetaInterface inputRowMeta, String name, RowMetaInterface[] info, StepMeta nextStep,VariableSpace space )

public void getFields( RowMetaInterface inputRowMeta, String name, RowMetaInterface[] info, StepMeta nextStep,VariableSpace space, Repository repository, IMetaStore metaStore )

注&＃xff1a;该方法在自己编写插件的时候&＃xff0c;并没有重写&＃xff0c;下一个组件就能直接拿到input的字段名

8.如果要添加新的或者修改一行数据&＃xff0c;可以用kettlt提供的&＃xff1a;RowDataUtil.addValueData来实现&＃xff1a;

Object[] outputRow &＃61; RowDataUtil.addValueData(r, data.outputRowMeta.size() - 1, "dummy value");

Object[] addValueData( Object[] rowData, int length, Object extra )

参数含义&＃xff1a;

rowData 构造出一行数据&＃xff0c;r &＃xff08;getRow(&＃xff09;获取&＃xff09;&＃xff0c;当前的行值。

length 要替换的值&＃xff0c;所在r的下标

extra 新的值

该类提供了下列方法:

9.关于数据分流&＃xff1a;

在processRow中使用&＃xff1a;

其中putError后面三个记录&＃xff0c;需要在*Meta中重写&＃xff0c;然后在图形化中鼠标右键&＃xff0c;定义错误数、错误列名等等&＃xff0c;就能在下个组件中拿到直接拿到&＃xff1a;

&＃64;Override

public boolean supportsErrorHandling() {

return true;//返回true即可。

}

整体实现&＃xff1a;一个自定义元数据&＃xff0c;根据元数据中定义的数据&＃xff0c;来检查input表中的数据是否满足&＃xff0c;

其中元数据索引表&＃xff0c;存储表、表字段、元数据值code。

根据元数据code查询出一个list集合&＃xff0c;该集合中的值&＃xff0c;为元数据索引表中指定字段允许的值。

最终插件实现效果&＃xff1a;