Kettle中Java节点生成

一、问题背景

现有的Transformer无法满足需求，需要购买第三方插件、开发插件、自定义Java类等办法，自定义Java类，因其代价小且门槛较低而成为最为常用的定制方法。

二、原理剖析

Kettle转换的核心对象/脚本类别，属于典型的需要编程基础才能掌控的Transformer类型。而Java代码Transformer，适用于熟悉Java语言的开发人员，用好这个Transformer，需要对类、接口、多线程等语言相关知识有所掌握，并且需要对Kettle的基础框架有所理解。

Kettle转换的执行，有以下三个核心的生命周期节点：

1、初始化

Kettle转换在执行前，会有一个各Transformer的初始化动作，为Transformer执行前的准备工作创造机会。为提高初始化的性能，Kettle为每个Transformer启用一个初始化线程，从而并行完成所有Transformer的初始化。初始化的主要内容就是调用一次Transformer的以下方法：

1	public boolean init( StepMetaInterface meta, StepDataInterfacedata)

此方法包含两个参数。其中，meta为元数据，data为数据。如果返回true，那么代表初始化成功，否则代表初始化失败。任何一个Transformer初始化失败，都会导致整个转换停止执行（在停止前，会调用每一个转换的资源释放方法dispose）。

2、执行

执行阶段是每一个Transformer实现特定价值的时候。为提高效率，Kettle为每一个Transformer单独启动一个工作线程来执行任务。Java程序员都了解，线程的核心代码是覆盖run方法。为简化起见，我将不重要的代码删除，得到工作线程run方法核心代码：

可以看出，线程一直在执行Transformer的processRow方法，直到出现以下情况之一：

· processRow方法返回false

· isStopped方法返回true

· processRow方法执行过程中出现异常，

其中，第一种情况代表工作已经正常完成；第二种情况，代表Transformer被强制停止；第三种情况，代表执行过程中出现错误，Kettle将调用stopAll方法，从而导致整个转换的所有工作线程停止执行。

执行方法的声明如下：

1	public boolean processRow( StepMetaInterface meta,StepDataInterfacedata ) throws KettleException;

每一个Transformer，都会在processRow方法中各显神通。一般的过程是，从输入行集中拿出一行，进行特定处理，然后将新的行放入输出行集中。从输入行集中取数据可以调用getRow方法。如果getRow方法返回值不为null，则Transformer应将该行数据进行处理，并调用putRow方法将处理结果存入输出行集，然后返回true，以继续为下一行输入数据处理提供机会。如果getRow方法返回null，代表输入行集已经处理完毕，这时可以调用setOutputDone，标识本Transformer执行完毕，并返回false，以结束本工作线程的执行。

3、资源释放

从上述工作线程的核心代码可以看出，不管工作线程是正常执行完毕还是异常执行完毕，最终会调用dispose方法。该方法声明如下：

1	public void dispose( StepMetaInterface meta, StepDataInterfacedata);

Transformer应该在需要时覆盖此方法，并释放相关资源。

了解上述原理后，撰写Java类Transformer中的代码时将胸有成竹。综上所述，一般情况下重写processRow方法即可满足需求，如果用到了一些重量级的资源，最好在init方法中初始化，并在dispose方法中释放。

由于Kettle使用Janino框架为自定义Java转换Transformer类动态定义了类名，并指定父类为TransformClassBase，所以在撰写代码时，只需要提供类的内容即可，无需class声明。

既然自动建立了父类，那么父类的成员、方法都可以在代码中重用。父类常用的成员包括以下三个实例：

parent：代表容器对象

meta：代表容器元数据对象

data：代表容器数据对象

常用的方法包括：

getRow：从输入行集中取一行数据

putRow：存银行数据到输出行集

stopAll：停止所有工作线程

setOutputDone：标记本Transformer工作完成

logBasic：输出基本日志

logError：输出错误日志

getInputRowMeta：得到输入行的元数据

createOutputRow：创建一个输出行数据

其实，常用的方法（如下图1所示），基本上都在Transformer属性对话框左侧Code Snippits中。一般情况下，可以双击其中的Main节点，从processRow方法的重写开始，需要其他代码时，在左侧找到对应代码块，双击即可加入。

三、案例分享

本文使用一个Kettle集成JMS的案例来进行实战演练。假设需要两个转换：一个转换名为Send，实现从文本文件输入流读取数据，并发送到ActiveMQ的队列；另外一个转换名为Receive，实现从队列读取数据，并发送到文本文件输出流。两个转换截图如下：

由于两个转换中用到的文本文件输入、输出都非常简单，这里只做简单描述。Send转换中，S01读取本地文本文件，包含两个String类型的字段ID、MENU_NAME。Receive转换中，S02输出文件到转换所在目录，仅包含一个名为MENU_NAME的String类型字段。

下文着重描述两个Java代码Transformer。第一个Transformer是Send转换中的S02，其主要代码注释如下：

第二个Transformer是Receive转换中的S01Transformer。主要代码注释如下：

注意，由于本文使用了ActiveMQ作为JMS服务器，所以为保证实例能够正常运行，需要自行下载服务器安装程序，并将对应jar文件拷贝到Kettle的lib目录下（本例使用activemq-all-5.8.0.jar）。代码中，需要的import指令如下：

import java.util. * ;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.MapMessage;
import javax.jms.MessageProducer;
import javax.jms.Session;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

四、总结

Guadazi

Kettle中的Java节点与代码生成

Kettle中Java节点生成