用 IBM WebSphere DataStage 进行数据整合：第 3 部分: 实时数据整合

 

【转载】用 IBM WebSphere DataStage 进行数据整合：第 3 部分: 实时数据整合

原文地址：http://www.ibm.com/developerworks/cn/data/library/techarticles/dm-0611zhoudp/index.html

 

本系列的第 1 部分和第 2 部分详细介绍了如何利用 DataStage 开发 ETL Job。 在本文中，我们将引入 RTI Job 的概念，并详细讲述如何利用 DataStage 开发 RTI Job 并将 RTI Job 发布成 Web Service 以供其他应用程序调用。

 

本文主要包含以下几部分内容：

1． RTI Job特性介绍

2． 开发一个RTI Job

3． 将RTI Job发布成Web Service

4． 利用Java客户端调用发布好的Web Service

RTI Job特性介绍

RTI Job是 ETL Job的一种特殊形式，它可以被发布成RTI 服务(RTI Service)，RTI服务分成三种，分别是Web service，EJB和JMS。在本文中我们只介绍如何发布成Web service。正是因为RTI Job可以以服务的形式被发布出来，这就使得企业数据整合和企业的面向服务的架构（SOA）无缝的整合在了一起。下面我们通过图一来看一下RTI Job被发布成RTI服务后整个应用程序的架构。

图一：RTI 架构
 

从这张图中，我们可以清楚地看出RTI的架构，首先开发好的RTI Job的实例运行在DataStage Server上，这些RTI Job以RTI Service的形式发布在RTI Server上。RTI Server和DataStage Server之间通过RTI Agent进行通信。其实RTI Server本身也是作为一个Web应用程序运行在Web应用程序服务器上的，可以是IBM Websphere 应用程序服务器或者BEA WebLogic应用程序服务器等。RTI Server和DataStage Server可以不放在同一台机器上，但是RTI Agent和DataStage Server必须放在同一台机器上面。这样，在RTI Server上部署好的Web Service就可以被各种客户端来调用，比如Java, dot NET等。

RTI Job的种类

RTI Job分成三种，第一种和普通的ETL Job没什么区别，既不含有RTI输入组件(RTI Input Stage)也不含有RTI 输出组件(RTI Output Stage)。第二种只含有RTI输出组件，第三种既含有RTI 输入组件又含有RTI输出组件。下面我们分别来介绍一下这三种类型的RTI Job

1． 既不含RTI输入组件也不含RTI输出组件

前面提到，这种RTI Job和普通的ETL Job没有什么区别，那么它怎么同调用它的客户端进行交互呢，这分两种情况，第一种情况是客户端只是希望调用一个ETL Job来执行数据的抽取，转换和加载的工作，不需要对该ETL Job有输入也无需从该ETL Job中得到输出。第二种情况是通过对该ETL Job设置参数来获得从客户端的输入，在这种情况下，该ETL Job有输入但没有输出。大家注意到，在这两种情况下，客户端都不能从该ETL Job中获得返回值，也就是说该ETL Job没有输出。为什么会这样呢，因为RTI Job若想有输出，那么它必须含有RTI 输出组件。图2显示了这种既不含RTI输入组件也不含RTI输出组件的RTI Job的一个例子。

图2：RTI Job示例1
 

2． 只含有RTI输出组件

这种RTI Job含有RTI输出组件但不含有RTI输入组件，这种RTI Job能通过RTI输出组件向调用它的客户端返回需要的值，可以是一个整型数据，一个结构化的数据或者一个结构化的数组，具体返回什么，可以在部署成Web Service的过程中设置，这会在文章的第三部分进行介绍。当然，这种类型的RIT Job也是通过设置Job的参数来接收从客户端传过来的参数的。图3显示了一个只含有RTI输出组件的一个例子。

图3：RTI Job示例2
 

3． 既含RTI输入组件又含RTI输出组件

这种RTI Job既能通过RTI输入组件接受从客户端传过来的参数，又能通过RTI输出组件来向客户端返回结果。这种RTI Job与前两种RTI Job的另外一个不同点是，当这种RTI Job编译后，会产生一个或者多个实例一直运行，这些事例会维持着和数据库的连接，因此当客户端进行调用的时候，相应速度就会非常快，而前面两种RTI Job的响应速度则相对会比较慢一些。图4是一个既含RTI输入组件又含RTI输出组件的RTI Job的例子。

图4：RTI Job示例3
 

在介绍了各种不同的RTI Job之后，我们接下来进入如何开发一个RTI Job的阶段。

开发一个RTI Job

在这一部分中，我们将详细讲解如何开发一个既含RTI输入组件又含RTI输出组件的RTI Job。首先我们来介绍一下我们即将开发的RTI Job的功能：我们的数据库系统中预先建立了一张表，表明为Student，这张表由两个字段，分别是ID和Name，我们即将开发的RTI Job就是通过接受从客户端传进来的ID来查找对应的Name，然后将查询结果返回到客户端去。

1． 打开DataStage Designer, 系统会自动弹出一个创建新的DataStage Job的对话框来，如图5所示，选择Server Job，然后单击OK按钮；

图5：新建Server Job
 

2． 这时候系统创建的Server Job如图6所示, 这时候这个Server Job上还没有放任何DataStage组件，注意我们现在还没有保存这个Server Job，单击左上角的保存按钮或者直接用Ctrl+S对Server Job进行保存；

图6：创建的Server Job
 

3． 如图7所示，在弹出的保存对话框中，输入Server Job的名称以及要保存的路径，在我们的例子中，我们以RTIJob为Server Job的名称，保存在文件夹RTISample中；

图7：保存Server Job
 

4． 保存好后，我们把需要的DataStage的组件从左边的组件面板拖入到右边的设计器中，如图8所示，我们需要一个RTI 输入组件，一个RTI输出组件，一个Transformer组件和一个DB2 UDB组件。选择好组件后，按照图8所示的逻辑进行连接，并分别为每个组件赋上一个有意义的名字；

图8：选择需要的组件
 

5． 在各个组件都准备好之后，我们需要对每个组件的属性进行设置。由于对属性进行设置的时候需要用到我们之前提到的Student表的表结构，我们首先需要从数据库中导入Student表的表结构，我们接下来的几个步骤就是进行表结构导入工作。在DataStage Designer中，如图9所示，右键单击Table Definition，然后选择Import ' Plug-in Meta Data Definition；

图9：导入表结构
 

6．如图10所示，在弹出的对话框中，选择DSDB2，然后单击OK按钮；

图10：选择数据库类型
 

7．然后在Server Name下拉列表中选择SAMPLE，我们创建的Student表正是存放在这个数据库中的，然后输入用户名和密码，并注意一定要选上Tables选择框，然后单击Next按钮，如图11所示；

图11：选择数据库连接信息
 

8．如图12所示，选择表Student，然后输入表结构要保存的路径，系统提供了一个默认的保存路径，我们可以修改成为我们想要保存的地方。然后单击Import按钮来导入表Student的表结构；

图12：选择要导入的表结构
 

9．接着我们来编辑DB2 UDB组件的属性，双击组件，如图13所示的属性设置框就会弹出来，在Stage标签的General子标签中，我们输入Server Name为Sample，然后输入连接数据库用的用户名和密码，在事务隔离级别(Transaction Isolation)的下拉框中，我们使用系统的默认值，游标稳定性隔离级别。然后单击Output标签；

图13：设置数据库连接信息
 

10．如图14所示，在Output标签的General子标签中，在Table Name输入框中输入表名Student，然后在Query Type下拉框中选择Generated SQL Query. 最后单击子标签Column进行进一步的属性设置；

图14：选择要连接的表
 

11．如图15所示，在Column子标签中单击按钮Load..，系统会弹出一个对话框让选择需要引入的表结构，我们选择刚才从数据库中导入的表Student的表结构，然后单击OK按钮；

图15：引入表结构
 

12．如图16所示，在弹出的对话框中可以选择需要导入表Student的那些列，系统默认是全部导入，保持默认，单击OK按钮；

图16：选择要导入的列
 

13．如图17所示，导入表结构后，Column标签里面会显示出我们从数据库中导入的表Student的表结构，单击View Data按钮来测试一下连接；

图17：导入后的表结构
 

14． 单击View Data按钮后，如果所有的设置都正确的话，系统就会弹出如图18所示的窗口，里面显示了Student表的内容，在我们的例子中，Student表里面有两条记录。至此，DUB UDB组件的属性设置已经完成。如果在这一步中不能成功显示表中的数据的话，就需要回过头来看看前面的设置是不是正确；

图18：测试连接
 

15．接下来我们设置RTI Input Stage的属性，双击RTI Input Stage组件，系统会弹出如图19所示的窗口，在Stage标签的General子标签中，你可以为该组件加一些注释以表明该组件的用途，也可以在这里设置组件的显示现实名称。然后我们单击Outputs标签；

图19：RTI输入组件属性设置
 

16．在Outputs标签的Columns子标签中，可以直接输入和Student表的ID字段相同的字段定义，也可以通过Load按钮来从Student表结构中只选择导入ID字段的定义。注意这里的ID字段正是需要从客户端接受的参数。当然客户端可以传过来一个ID的值或者以数组的形式传过来多个ID的值，具体的设置是在部署Web Service的过程中进行的。设置完成后，单击OK按钮，如图20所示；

图20：引入字段定义
 

17．然后我们对RTI Output Stage组件属性进行设置，双击RTI Output Stage组件，如图21所示的窗口会显示出来，像对RTI Input Stage组件进行属性设置一样，这里同样可以加入一些描述信息，更改RTI Output Stage组件的显示名称等，我们保持默认，然后单击Inputs标签；

 

18．在Inputs标签的Columns子标签中，直接输入如图22所示的内容，当然，你也可以从Student表中只导入Name字段的定义，注意我们这里的Name字段正是需要返回给客户端的数据，当然是返回一个还是多个值是通过在部署Web Service的过程中来设置的。最后单击OK按钮来完成设置；

 

19．然后我们来设置Transformer Stage组件的信息，具体设置的内容如图23所示。在该图中，左上角的Parameter框代表的是RTI Input Stage组件，它只含有一个字段ID，左下角的Data框代表的是DB2 UDB组件，连接的是Student表，它有两个字段ID和Name。由下角的Name框代表的是RTI Output Stage组件的信息，它只有一个字段Name，这也正是需要返回给客户端的信息；

图23：配置Transformer Stage组件
 

20．配置完每个组件的属性后，我们需要对整个Job的属性进行配置，以使得DataStage系统知道这是一个RTI Job。单击DataStage Designer工具栏中的Job Properties图标，如图24所示的对话框就会弹出来，然后选上Allow Mutiple Instance和RTI Service Enabled选择框。这样就能利用DataStage提供的RTI控制台将该RTI Job发布成Web Service。最后单击OK按钮并保存更改；

图24：配置RTI Job
 

21．这时候我们已经配置好了所有需要配置的信息，接下来我们来编译开发好的RTI Job，单击DataStage Designer工具栏中的编译图标，编译成功后的画面如图25所示。与一般的DataStage Job不同的是，如果你开发的是一个既含有RTI输入组件又含有RTI输出组件的RTI Job，那么在编译后就会自动运行这个RTI Job的实例。我们可以通过DataStage Director来观察；

图25：编译RTI Job
 

22．如图26所示，打开DataStage Director，在RTI Sample目录下面，你会看到一个RTI Job的实例正在运行，这个实例维持着与数据库的连接，如果你把数据库连接断掉，你会发现这个实例就会中断掉，然后系统会接着尝试启动一个新的实例，直到数据库连接恢复正常；

图26：查看RTI Job的状态
 

现在我们已经成功开发并编译了RTI Job，我们接下来的工作就是将这个RTI Job发布成Web Service。

将RTI Job发布成Web Service

这一部分我们将详细讲述如何将一个已经编译好的RTI Job发布成Web Service，在这一部分进行之前，你需要检查一下你的软件的安装，确保以下软件已经成功安装好：

1． RTI Console 它是将RTI Job部署成Web Service的可视化接口，并且提供了部署的向导方便我们进行部署

2． RTI Agent 要确保它已经同DataStage Server安装在了同一台机器上，RTI Agent的安装一般是和DataStage Server的安装一起安装的，RTI Agent负责RTI Server和DataStage Server的通信。

3． RTI Server 部署的Web Service是运行在RTI Server上的，当然RTI Server也是运行在一个Web应用程序服务器上的，比如IBM Web Application Server。

准备好需要的软件后，我们进入利用RTI控制台发布Web Service的部分。

1．首先打开RTI控制台，如图27所示，DataStage提供了一个图形化的界面方便我们把RTI Job发布成RTI Service。然后我们在左侧面板的Current Tasks模块中单击链接Register an RTI Server来新注册一个RTI Server；

图27：RTI 控制台
 

2．在弹出的对话框中，输入RTI服务器的名字或者IP地址，然后在Web Service End-point中输入如图28所示的值，这里要注意，因为我是把RTI Server部署在IBM Websphere 应用程序服务器上的，而运行在其上的应用程序都是运行在9080端口下面，所以我设置的端口号是9080，如果你是将RTI Server运行在其他的应用程序服务器上，你就需要将这个端口号改成该应用程序服务器所对应的端口号，最后单击Finish按钮；

图28：注册RTI Server
 

3． 如图29所示，我们会看到右边的面板中出现了我们刚才注册好的RTI Server，双击这个图标。

图29：成功注册RTI Server
 

4． 如图30所示，在双击刚才注册的RTI Server之后，左边的菜单选项也发生了相应的改变，我们在Current Tasks模块中单击链接Register a DataStage Machine来注册一个DataStage服务器。

图30：注册DataStage服务器
 

4． 弹出的对话框如图31所示，我们可以在Machine Name输入框中输入DataStage服务器所在的计算机名或者IP地址。如果需要的话，我们还要输入登陆的用户名和密码，其他选项保持默认，然后单击Finish按钮。

图31：填写注册信息
 

5． 如图32所示，注册好的DataStage服务器会显示在右边面板的上半部分中，注册成功后，我们单击左侧菜单Current Tasks模块中的链接Add a new Service to the RTI Server去注册一个新的RTI Service；

图32：注册新的RTI Service
 

6．如图33所示，在弹出的对话框中，在Service Name文本框中输入sample，其他的内容保持默认，然后单击Finish按钮；

图33：填写配置信息
 

6． 如图34所示，注册好的RTI Service显示在右边面板的下半部分，注意我们现在注册的这个RTI Service只是一个空的壳子，我们还需要为它选择使用的协议以及附加的RTI Job，这些都会在下面的步骤中进行配置。最后我们双击这个RTI Service的图标；

图34：注册好的RTI Service
 

8．如图35所示，单击左侧面板Current Tasks模块中的链接Add support for Service Bindings；

图35：添加Service Binding
 

9．如图36所示，选择Service Binding的类型为SOAP over HTTP，因为我们是将RTI Job发布成Web Service，所以我们必须选择这个选项，其他选项是用来把RTI Job发布成JMS或者EJB,然后单击Next按钮；

 

10．如图37所示，你可以输入一些描述信息，然后单击Finish按钮；

图37：添加描述信息
 

11．如图38所示，我们添加的Service Binding显示在了右侧面板的上半部分中，单击该Service Binding的图标，然后从左边面板的Current Tasks模块中单击链接Add an operation来为该RTI Service添加RTI Job；

图38：注册好的Service Binding
 

12．如图39所示，在弹出的对话框中选中我们之前开发并编译好的RTI Job RTIJob，然后单击按钮Next；

图39：选择RTI Job
 

13． 如下图所示，我们需要在Operation Name文本框中输入该RTI Job所对应到Web Service中的名字，直接输入RTIJob，保持其他选项为默认，然后单击按钮Next；

图40：配置相关信息
 

14．在如图41所示的页面中，我们可以配置该Web Service所接受的参数的信息，注意系统已经默认为我们生成了配置，那就是只接受一个名称为ID参数，类型为String，我们可以通过Options下来列表来改表输入参数的设置，比如如果我们需要传入的参数是结构化的，那么我们需要选择选项Group，如果我们需要传入多个相同类型的数据，那么我们的参数就需要设置成Array，在我们的例子中，如果我们设置成了Array，那么就代表我们的Web Service会接受一个字符串数组，里面可能存放了多个人的ID信息。由于我们这个例子只是传入一个ID，然后返回该ID所对应的Name，所以我们不需要进行设置，保持系统默认，然后单击按钮Next。

图41：配置输入参数
 

15．接着我们需要配置该Web Service向客户端的返回值，由于我们只需要返回一个字符串类型的数据，我们保持系统的默认设置，然后单击按钮Next；

图42：配置返回值类型
 

16．在这里可以设置一些和WSDL文件相关的名字空间的信息，我们使用系统提供的默认值，然后单击Next按钮；

图43：名字空间的设置
 

17．如图44所示，确认选择框Launch the Add Job/Map Wizard with the job or map you selected是选上的，然后单击按钮Finish；

图44：完成配置
 

18．在图45中可以配置一些运行时的参数，比如RTI Job的实例的最小数目是多少，最大树木是多少等，可以根据自己的应用程序的需要来配置，比如如果同时会有多个客户端来访问该Web Service，那么就需要将最小实例数目设置大一些。 设置好后，单击按钮Next；

图45：配置运行参数
 

19．在下图中输入验证信息，然后单击按钮Finish；

图46：输入验证信息
 

20．如下图所示，系统弹出一个对话框提示用户已经成功的将RTI Job添加到了RTI Service当中；

图47：添加成功
 

21．如图48所示，我们需要重新激活我们之前配置的Service Binding，并且我们注意到我们添加的RTIJob也显示在了右边面板的下半部分；

图48：激活Service Binding
 

22．单击左侧面板的View RTI Service Repository链接，就进入到了如图49所示的页面,然后单击我们刚才注册成功的Web Service sample；

图49：部署的RTI Service
 

23．下图显示了我们刚部署成功的Web Service的相关信息，我们单击对应的WSDL文件的链接去查看WSDL文件；

图50：Web Service 相关信息
 

24．图51显示了我们部署好的Web Service对应的WSDL文件的内容，而各种客户端正是通过这个WSDL文件对后台的应用逻辑进行调用的。我们在文章的最后一个章节会介绍如何调用该Web Service；

图51：WSDL文件
 

利用Java客户端调用发布好的Web Service 文章接下来讲述如何去调用我们之前发布好的Web Service，我们使用一个用Java语言编写的客户端来进行调用。在开始之前，读者应该准备好如下的环境：

1．Eclipse集成开发环境

因为我们开发的Java项目是在Eclipse下面进行的，所以安装一个Eclipse的环境能使读者更容易的跟上文章的每一步。可以从Eclipse的官方网站上下载最新版本的Eclipse。

2．JDK1.4或者1.5

这是Java开发的必备环境。

3． Axis

文章中会使用Axis提供的工具生成调用Web Service的本地Stub，方便对Web Service进行调用。Axis可以从它的官方网站下载到，文章后面的资源列表包含了该链接。

下面我们开始Java客户端的开发

1． 首先使用创建一个Java工程，命名为TestRTIJob，然后右键单击这个工程并选择属性，会弹出如图52所示的窗口，单击Library标签，然后把下载的Axis的JAR文件加入进来。

图52：添加外部JAR文件
 

2． 然后从Eclipse中选择Run ' Run..，会弹出如下图所示的对话框，选择程序运行的类别为Java Application，然后单击按钮Search；

图53：新建一个运行实例
 

3．如下图所示，从弹出的对话框中选择类WSDL2Java，这个类是Axis提供的用来从WSDL生成本地Stub类的工具。然后单击OK按钮；

图54：选择运行主类
 

3． 然后在Argument标签中输入我们刚才部署好的Web Service的WSDL文件的地址，如图55所示，然后单击按钮Run；

图55：运行WSDL2Java
 

4． 如图56所示，运行结束后，我们会看到我们创建的工程里面生成了一些类，这些类就是辅助我们调用Web Service的Stub类；

图56：生成的Stub类
 

5．我们接下来新建一个叫做TestRTIJob类来通过生成的Stub类来进行调用Web Service，这个类的源代码如下所示：

清单1：调用Web Service

package com.ascential.rti.sample;
            import java.rmi.RemoteException;
            import javax.xml.rpc.ServiceException;
            public class TestRTIJob {
            public static void main(String[] args){
            SampleLocator locator = new SampleLocator();
            SampleDOCLIT service = null;
            try {
            service = locator.getsampleSoap();
            String name = service.RTIJob("001");
            System.out.println("The name is: " + name);
            } catch (ServiceException e) {
            e.printStackTrace();
            } catch (RemoteException e) {
            e.printStackTrace();
            }
            }
            }
            

运行这个类，控制台会打印出一串字符串：The name is Jason，这说明我们成功的调用了Web Service，我们的RTI Job也成功的完成了我们所设计的业务逻辑。关于这个Java工程的详细信息，读者可以在文章的下载部分下在提供的整个工程的源代码。

至此，我们已经完成了从开发RTI Job到部署RTI Job再到用Java Client调用部署的Web Service的整个过程。

总结

本文首先介绍了什么是RTI Job以及RTI Job的特点，然后通过开发、部署RTI Job使读者对RTI Job有了更进一步的了解，最后通过开发一个Java客户端对Web Service进行调用来结束了整个文章的讲述。在面向服务的架构日益成熟的今天，通过把ETL Job发布成Web Service供企业的其他应用程序进行调用无疑会是一个非常有价值的工作，从而使企业数据整合与企业的面向服务的架构实现了无缝的融合。希望本片文章对使用DataStage进行数据整合的读者有所帮助。

参考资料

1. 了解关于DataStage产品的更多信息：http://ibm.ascential.com/
2. Eclipse官方网站：http://www.eclipse.org/
3. 下载最新版本的JDK：http://java.sun.com/javase/downloads/index.jsp
4. Axis官方网站：http://ws.apache.org/axis/