primetondi6 2userdocdi6 2ga高级开发指南

1、PRIMETON TECHNOLOGIES, LTD.普元信息技术股份DI 6.2 GA高级开发指南No part of this document may be reproduced, stored in anyelectronic retrievalsystem, orotherwise,transmitted in any form or by any means, mechanical, photocopying, recording,without the written permission of the copyright owner.COPYRIGHT 2015 by Prim

2、eton Technologies, Ltd. ALL RIGHTS RESERVED./第 1 页 共 57 页 DI6.2 GA 高级开发指南目录1概述41.1目标41.2前提4运行机制42.1 模块关系42.2 元数据52.2.1 数据库连接元数据62.2.2 业务模型元数据72.2.3 任务调度元数据72.3 数据处理机制82开发环境准备10资源库调用104.1 文件资源库104.2 数据库资源库12远程调用13组件开发146.1包路径176.2开发示例176.2.1步骤 1186.2.2步骤 2186.2.3步骤 3226.2.4步骤 42

3、46.2.5步骤 5256.2.6步骤 6296.2.7步骤 730模型编码实现31扩展开发参考373456788.1JMX API378.1.1 DataIntegrationBZMBean 类378.1.2 DataIntegrationProcessMBean 类388.1.3 DataIntegrationProcessMBean 类388.2 扩展 API388.2.1 StepExecutionStatus 组件状态分类388.2.2 Trans 业务模型398.2.3 Result 运行结果集418.2.4 TransMeta 业务模型元数据428.2.5 TransHopMet

4、a 连线元数据43/第 2 页 共 57 页 DI6.2 GA 高级开发指南8.2.6 RowMetaInterface 字段列表元数据448.2.7 ValueMetaInterface 字段元数据468.2.8 Job 任务调度478.2.9 JobMeta 任务调度元数据488.2.10 Repository 资源库508.3组件开发接口51StepMetaInterface 元数据接口51StepDataInterface 数据接口52StepInterface 组件运行接口52StepDialogInterface 界面接口54TRANS

5、日志表54转换日志表54步骤日志表54步骤运行日志表55日志通道日志表55度量日志表56JOB 日志表56作业日志表56作业项日志表56日志通道日志表5.....3/第 3 页 共 57 页 DI6.2 GA 高级开发指南1 概述1.1 目标本文档主要对开发人员在编写 DI 功能扩展时提供参考，文档描述了 DI 各个组成部分之间的调用关系和模型运行机制，提供各种 API 开发时所需要的接口说明和典型开发示例。 1.2 前提在阅读本

6、文当前需要能够熟练操作 DI 的各个模块，了解 Eclipse 下 java 编程基础才能 进行扩展开发。运行机制2本章主要描述 DI 各个模块之间的关系，以及 DI 是如何基于元数据运行的。另外描述DI 模型在Server 中的运行机制。 2.1 模块关系DI 由以下六部分产品组件组成：DI Server（数据集成引擎）、DI Component Library（数 据集成组件库）、DI Studio（数据集成建模工具）、DI Governor（数据集成模型治理）、DIMetadata Warehouse（元数据仓库）、DI Agent（前置节点）。 http:/www.primeton.c

7、om/第 4 页 共 57 页 DI6.2 GA 高级开发指南这几个部分在Governor 和Server 之间都是通过Server 或者Agent 发布的JMX 来通信的。 在 Studio 中建好的模型导出文件。通过在 Governor 中配置作业，在部署的时候会调用 作业配置中选择的 Server 服务器的信息，通过 JMX 调用 Server 或 Agent 发布的部署作业接 口来部署作业模型。在配置好调度启动运行调度的时候也会通过JMX 调用 Server 发布的运行作业的接口来运行作业。在扩展开发中用户可以根据需求和实际情况编写JMX 客户端来调用 Server 或 Agent 发

8、 布的各个接口来扩展 Server 的功能。 2.2 元数据DI 的元数据（Metadata）是关于数据、操纵数据的进程，以及应用程序的结构、意义的 描述信息，其主要目标是提供数据资源的全面指南。按用途分为以下几类：技术元数据数据源信息、连接池信息、表结构、字段属性等。/第 5 页 共 57 页 DI6.2 GA 高级开发指南业务元数据转换规则、映射规则、合并规则、业务处理脚本等。技术元数据与业务元数据的统一存储保证了 ETL 流程的可配置、可管理、可优化，同时便于明确业务数据的具体来源。2.2.1数据库连接元数据数据库连接信息在数据库中的存储结构：

9、表结构说明：/第 6 页 共 57 页 表名 元数据库描述 R_DATABASE存放资源库连接信息 R_DATABASE_CONTYPE存放DI 支持的数据库连接类型 R_DATABASE_TYPE存放DI 支持的数据库类型 DI6.2 GA 高级开发指南2.2.2业务模型元数据业务模型元数据信息在数据库中的存储结构：表结构说明：2.2.3任务调度元数据任务调度元数据信息在数据库中的存储结构：/第 7 页 共 57 页 表名 元数据库描述 R_TRANSFORMATION存放业务模型信息 R_TRANS_A

10、TTRIBUTE存放业务模型中的配置属性信息 R_TRANS_HOP存放业务模型的连线信息 R_TRANS_NOTE存放业务模型中的注释 R_STEP存放业务模型中的组件信息 R_TRANS_STEP_CONDITION存放业务模型中组件的条件信息 DI6.2 GA 高级开发指南表结构说明：2.3 数据处理机制DI 基于数据流的数据处理模型，能最大程度的发挥各步骤的处理能力。同时支持多实 例并行执行。/第 8 页 共 57 页 表名 元数据库描述 R_JOB存放任务调度模型信息 R_JOBENTRY存放任务调度模型中的组件信息 R_JOB_NOTE存

11、放任务调度模型的注释 R_JOBENTRY_COPY存放任务调度模型组件位置信息 R_JOBENTRY_TYPE存放任务调度的组件 R_JOB_HOP存放任务调度模型的连线信息 DI6.2 GA 高级开发指南模型在运行的时候会为每个组件分配单独的运行线程并启动组件运行类。组件以行为单 位处理，其中行的元数据信息 RowMeta 和数据信息统一保存在 RowSet 对象中。这样每一个组件在获取数据的时候都会从 RowSet 缓存里获取数据进行交互。 组件与组件之间以行为单位进行处理，自然需要知道每行的结构，即行元数据。行元数 据至少需要包括类型、名称，当然还可能包括字段长度、精度等常见内容。行元

12、数据不仅在执行的时候需要，而且在转换设置的时候同样需要。每个步骤的行元数 据都会保存在.dbr 文件或者数据库里面，所以可以根据步骤名称从 TransMeta 对象中获取行 元数据。数据在从一个步骤流转到下一个步骤的时候会将数据先放入一个RowSet 的缓存中。 RowSet 的实现类，负责步骤之间的相互通信，rowset 对象既是前一个 step 的成员也是 后一个 step 的成员，访问是线程安全的。 RowSet 类中包含源 step，目标 step 和由源向目标发送的一个 rowMeta 和一组 data。其 中 data 数据是以行为单位的队列（queArray）。一个 RowSet

13、 作为此源 step 的 outputrowsets的一部分。同时作为目标 step 的 inputRowsets 一部分。源 Step 每次向队列中写一行数据， 目标 step 每次从队列中读取一行数据。 /第 9 页 共 57 页 DI6.2 GA 高级开发指南3开发环境准备在开发扩展功能前需要先准备好开发所需要的JDK 和 Eclipse。启动 Eclipse 后，新建一 个 java 工程。然后给工程添加扩展开发所需要的包。扩展开发所需要的包主要都在 studio 根目录下的 plugins 和 libs 目录下，将这两个目录下的 jar 包

14、添加到构建路径中。 4资源库调用本节主要以示例的方式讲述如何使用 API 连接 DI 所提供的两种资源库。资源库的主要 作用就是保存用户在 Studio 中创建的模型信息。 4.1 文件资源库文件资源库是一种常用的资源库形式，它将模型保存成dbr 和 dsr 两种类型的文件，分别表示业务模型和业务调度。接下来我们来看看如何使用 API 来连接并使用这种资源库。 /第 10 页 共 57 页 DI6.2 GA 高级开发指南/第 11 页 共 57 页 import com.primeton.di.core.D

15、ataIntegrationEnvironment;import com.primeton.di.core.exception.DataIntegrationException;import com.primeton.di.repository.filerep.DataIntegrationFileRepository; import com.primeton.di.repository.filerep.DataIntegrationFileRepositoryMeta; import com.primeton.di.trans.Trans;import com.primeton.di.tra

16、ns.TransMeta;publicclass ReaderTransFromRep privatestatic String tranName = IM_KSDM; / 传输名称 publicstaticvoid main(String args) throws DataIntegrationException Trans trans = null; try / 初始化 DataIntegrationEnvironment.init();/ 这个是4.0版本的初始化，和3.x版本的不同 / 资源库元对象 DataIntegrationFileRepositoryMeta repinfo =

17、 new DataIntegrationFileRepositoryMeta( , , 数据采集, file:/E:/Workspaces/DataIntegration);/ 文件形式的资源库 DataIntegrationFileRepository rep = newDataIntegrationFileRepository(); rep.init(repinfo);/ 转换元对象 if (tranName != null) TransMeta transMetadel =rep.loadTransformation( rep.getTransformationI D(tranName,

18、 null), null);/ 转换 trans = new Trans(transMetadel);/ 执行转换 trans.execute(null);/ 等待转换执行结束 trans.waitUntilFinished();/ 抛出异常 if (trans.getErrors() 0) thrownew Exception(There are errors during transformation exception!(传输过程中发生异 常); else thrownew DataIntegrationException(传输名为空！); catch (Exception e) if

19、(trans != null) trans.stopAll();DI6.2 GA 高级开发指南4.2 数据库资源库数据库资源库将模型保存到数据库的数据表中，接下来我们来看看如何使用 API 来连 接并使用这种资源库。/第 12 页 共 57 页 import com.primeton.di.core.DataIntegrationEnvironment;import com.primeton.di.core.database.DatabaseMeta;import com.primeton.di.core.exception.DataIntegrat

20、ionException;import com.primeton.di.repository.RepositoryDirectory;import com.primeton.di.repository.RepositoryDirectoryInterface;import com.primeton.di.repository.kdr.DataIntegrationDatabaseRepository; import com.primeton.di.repository.kdr.DataIntegrationDatabaseRepositoryMeta; import com.primeton.

21、di.trans.Trans;import com.primeton.di.trans.TransMeta;publicclass ReaderTransFromDBRep privatestatic String transName = t1; publicstaticvoid main(String args) try DataIntegrationEnvironment.init();DatabaseMeta dataMeta = new DatabaseMeta(DataIntegrationDBRep, MSSQL, Native, , etl, 1433, sa,

22、000000);DataIntegrationDatabaseRepositoryMetarepInfo=newDataIntegrationDatabaseRepositoryMeta();repInfo.setConnection(dataMeta);DataIntegrationDatabaseRepository rep = new DataIntegrationDatabaseRepository(); rep.init(repInfo);rep.connect(admin, admin);RepositoryDirectoryInterface dir = new Reposito

23、ryDirectory(); dir.setObjectId(rep.getRootDirectoryID();TransMeta tranMeta =rep.loadTransformation( rep.getTransformationI D(transName, dir), null);Trans trans = new Trans(tranMeta); trans.execute(null);trans.waitUntilFinished();thrownew DataIntegrationException(e);DI6.2 GA 高级开发指南5远程调用DI Server 在启动后

24、会对外提供调用内部方法的 JMX 接口(详细接口内容请参考扩展开 发参考)。用户可以通过编写 JMX 客户端来调用 Server 提供的接口，控制模型的运行。下面 我们编写一个运行模型的示例。/第 13 页 共 57 页 package com.primeton.di.runtime.test;import java.io.IOException;import .MalformedURLException;import javax.management.InstanceNotFoundException;import javax.ma

25、nagement.MBeanException;import javax.management.MBeanServerConnection;import javax.management.MalformedObjectNameException;import javax.management.ObjectName;import javax.management.ReflectionException;import javax.management.remote.JMXConnector; import javax.management.remote.JMXConnectorFactory; i

26、mport javax.management.remote.JMXServiceURL; import com.primeton.di.runtime.jmx.JMXService;publicclass DIJMXClient / JMX域名 publicstaticfinal String JMXSERVICE_DOMAIN_NAME = JMXAgent;/ MBean显示名 publicstaticfinal String MBEAN_Name = MBean Name; publicstaticfinal String BZMBEAN_NAME = DataServiceBusine

27、ss; publicstaticfinal String TASKMBEAN_NAME = DataServiceTask; publicstaticfinal String PROCESSMBEAN_NAME = DSProcess;publicstaticvoid main(String args) String Port = 9999;/主机端口 String IP = ;/主机地址 try /建立JMX连接 JMXServiceURL jmxURL = new JMXServiceURL(service:jmx:rmi:/jndi/rmi:/ + IP+ : + Po

28、rt + /jmxServer);JMXConnector jmxConnector = JMXConnectorFactory.connect(jmxURL); MBeanServerConnection mbServerConnection =DI6.2 GA 高级开发指南6组件开发在 DI 中的组件根据执行的功能分为两种类型：即业务模型组件和任务调度组件。 业务模型组件：业务模型提供了一系列组件，完成了对数据的抽取、转换与加载 的操作，支持Oracle、DB2、MSSQL、MYSQL 等多种数据库以及文本、XML 等 多种数据源的读写。提供了一套功能强大的转换组件，以拖拽的方式完成数据映

29、 射、清洗、拆分等一系列复杂操作。任务调度组件：任务调度完成了对业务模型的调度工作，提供定时、验证、脚本、 协议等一系列组件可灵活的运行业务模型。DI 为扩展插件提供了 4 个扩展点: StepMetaInterface 元数据接口、StepDataInterface 数据 接口、StepInterface 运行接口和 StepDialogInterface 界面接口，这 4 个扩展点也是每个步骤的组成。每个类都有其特定的目的及扮演的角色。下面将以 ArcSDEInput 为例说明这 4 个类 /第 14 页 共 57 页 /创建MBeanObjec

30、tName bzObjName = new ObjectName(JMXSERVICE_DOMAIN_NAME,MBEAN_Name, BZMBEAN_NAME);ObjectName taskObjName = new ObjectName(JMXSERVICE_DOMAIN_NAME,MBEAN_Name, TASKMBEAN_NAME);ObjectName processObjName = new ObjectName(JMXService.JMXSERVICE_DOMAIN_NAME, JMXService.MBEAN_NAME, JMXService.PROCESSMBEAN_NA

31、ME);/运行流程转换1是模型文件名称mbServerConnection.invoke(bzObjName, execBusinessProcess,new Object 转换1,new String0,new String java.lang.String, java.lang.Object); catch (MalformedURLException e) e.printStackTrace(); catch (IOException e) e.printStackTrace(); catch (MalformedObjectNameException e) e.printStackTr

32、ace(); catch (InstanceNotFoundException e) e.printStackTrace(); catch (MBeanException e) e.printStackTrace(); catch (ReflectionException e) e.printStackTrace();DI6.2 GA 高级开发指南的继承体系。1.StepInterface实现 StepInterface 接口的类，在转换运行时，将是数据实际处理的位置。每个执行线程 都表示一个实现 StepInterface 的实例。BaseStep 实现了 StepInterface，是各

33、step 具体实现类的基类。完成了公用的处理函数，如 putRow()，但是对于更具体的 processRow()在 BaseStep 的子类中。BaseStep 的主要成员 有：public ArrayList inputRowSets，outputRowSets； BaseStep 的子类每次从 inputRowSets 中取出一行数据，向 outputRowSets 中写入一行数 据。2.StepDataInterface/第 15 页 共 57 页 DI6.2 GA 高级开发指南实现 StepDataInterface 接口的类为数据类，当插

34、件执行时，对于每个执行的线程都是唯 一的。保存step 相关的数据信息，比如行的元数据信息。 3.StepMetaInterface实现了 StepMetaInterface 接口的类为元数据类。它的职责是保存和序列化特定步骤的实 例配置，例如保存步骤的名称、字段名称等，如何生成加载 xml 或者读写数据库。 4.StepDialogInterface/第 16 页 共 57 页 DI6.2 GA 高级开发指南实现了 StepDialogInterface 接口的类为对话框类，该类实现了该步骤与用户交互的界面， 它显示一对话框，通过对话框用户可以根据

35、自己的要求进行步骤的设定。该对话框类与元数 据类关系非常密切，对话框里面的配置数据均会保存在元数据类里面。6.1 包路径DI 中已经将组件的功能开发和组件的界面开发完全分离。默认包路径名称中必须有 di关键字。在 DI 查找功能类时会沿着原有路径查找相关的类，在查找界面类时会将原有路径 中的 di 解析为 di.ui 包路径去查找这个包下的界面类。 例如：在 查 找 组 件 “ 表 输 出 ” 时 ， 会 将 指 定 的 组 件 类 路 径 “com.primeton.di.business.steps.arcsdeinput”解析为“com.primeton.di.ui.business.s

36、teps.arcsdeinput”路径，到这个路径下找到界面类。6.2 开发示例这里将新建一个用于接入GIS 数据库的输入组件来详细讲解组件开发的过程。 地理信息系统（Geographic Information System 或 GeoInformation system，GIS）有时 又称为“地学信息系统”或“资源与环境信息系统”。它是一种特定的十分重要的空间信息系统。 它是在计算机硬、软件系统支持下，对整个或部分地球表层（包括大气层）空间中的有关地 /第 17 页 共 57 页 DI6.2 GA 高级开发指南理分布数据进行采集、储存、管理、运算

37、、分析、显示和描述的技术系统。GIS 空间数据库：也就是地理信息系统数据库，或地理数据库。是某一区域内关于一定 地理要素特征的数据集合，为GIS 提供空间数据的存储和管理方法。 6.2.1步骤 1首先我们来创建数据库所需要基本的类，如下图所示：ArcSDEInput 类继承 BaseStep 类实现 StepInterface 接口。这个类主要负责处理数据将空 间数据从地理数据库中读取出来，然后放入下一个步骤。ArcSDEInputData 类继承 BaseStepData 类实现 StepDataInterface 接口。这个类负责存储 一些ArcSDEInput 运行时的变量，例如：字段类

38、型数组，数据库连接等。 ArcSDEInputMeta 类继承BaseStepMeta 实现 StepMetaInterface 接口。这个类主要用于存 储界面设置的元数据信息， 可以将设置的信息保存到数据库或文件中， 并能够为 ArcSDEInput 类提供所需要的元数据。这里将这些类创建好并实现里面的方法。后面我们将详细介绍类里面方法的具体内容。Messages 类为读取国际化文字类，这里不做过多介绍。6.2.2步骤 2这一步我们来编写一个设置空间数据库的界面，需要编写一个界面类，如下图：ArcSDEInputDialog 类就是空间数据库的界面类继承了 BaseStepDialog 类实

39、现了 StepDialogInterface 接口。我们先来看一下实现后的界面，如下图：/第 18 页 共 57 页 DI6.2 GA 高级开发指南首先在这个类中需要编写构造方法：这个构造方法中第一个参数是 SWT 的 Shell；第二个参数是一个 Object 类型的数据其 实就是步骤 1 中实现 BaseStepMeta 的类，因为后面保存和获取元数据需要用到这个类，所 以这里留了一个这个类的引用；第三个参数是业务模型的元数据类；第四个是这个组件的名 称，由最后给这个类变得的配置文件指定。由于需要从 ArcSDEInputMeta 类中获取和保存

40、配置信息，所以在Meta 类中增加界面配置参数对应的变量。例如：/第 19 页 共 57 页 private String username;/用户名 private String password;/ private String server;/服务器 public ArcSDEInputDialog(Shell parent, Object in, BusinessMeta businessMeta, String sname) super(parent, (BaseStepMeta)in, businessMeta, sname); inpu

41、t=(ArcSDEInputMeta)in;DI6.2 GA 高级开发指南StepDialogInterface 接口中需要实现 open 和 setRepository 这两个方法，setRepository 方 法在 BaseStepDialog 类中已经实现，open 方法中主要编写的是界面的展现，我们来看里面 实现的主要内容：接下来需要编写获取元数据的方法用于初始化界面，方法如下：/第 20 页 共 57 页 publicvoid getData()if(input.getServer()!=null)arcSDEServer.setText

42、(input.getServer();/从元数据中获取配置信息 if(input.getUsername()!=null) userText.setText(input.getUsernam e();.if (input.getColumns()!=null)String columns=input.getColumns().split(,);for (int i=0;i0) meta.setUsername(userText.get Text().trim();if(passwordText.getText().trim().length()0) meta.setPassword(passw

43、or dText.getText().trim();if(limitText.getText().trim().length()0) meta.setLimit(Integer.parseInt( limitText.getText().trim();.if(portText.getText().trim().length()0) meta.setInstance(Integer.parseInt(port Text.getText().trim();if(isStringButton.getSelection() input.setIsString(1);elseinput.setIsStr

44、ing(0);TableItem item = selectColumnList.table.getItem(i);if (columnsi!=null) item.setText(1, columnsi);if(input.getInstance()!=0) portText.setText(input.getInstance()+);if(input.getIsString()!=0) isStringButt on.setSelection(true);elseisStringButton.setSelection(false);DI6.2 GA 高级开发指南6.2.3步骤 3接下来需要

45、编写 ArcSDEInputMeta 类中的保存和读取配置信息的方法。先来编写设置 默认配置参数方法：然后是从资源文件中解析文件内容的方法：/第 22 页 共 57 页 /* 文件载入 */publicvoid loadXML(Node stepnode, List databases,Map counters) throws DataIntegrationXMLException tryusername = XMLHandler.getTagValue(stepnode, username); password = XMLHandler.getTa

46、gValue(stepnode, password); server = XMLHandler.getTagValue(stepnode, server);databaseName = XMLHandler.getTagValue(stepnode, databaseName); where = XMLHandler.getTagValue(stepnode, where);columns = XMLHandler.getTagValue(stepnode, columns); tableName = XMLHandler.getTagValue(stepnode, tableName); l

47、imit = Const.toInt(XMLHandler.getTagValue(stepnode, limit), 0);.catch(Exception e)publicvoid setDefault() username=; password=; server=localhost; databaseName=; where=; columns=; tableName=; limit=0;if (Const.isEmpty(wStepname.getText() return; stepname = wStepname.getText(); / return valuegetInfo(i

48、nput);/保存配置信息dispose();/关闭界面 DI6.2 GA 高级开发指南下面是从保存配置信息到文件的方法：编写完保存和读取文件的方法后还需要编写保存和读取数据库资源的方法。下面是保存 数据库资源的方法：/第 23 页 共 57 页 /* 将配置信息写入数据库 */publicvoid saveRep(Repository rep, long id_transformation, long id_step)throws DataIntegrationException tryrep.saveStepAttribute(id_transf

49、ormation, id_step, username, username); rep.saveStepAttribute(id_transformation, id_step, password,password); rep.saveStepAttribute(id_transformation, id_step, server,server); rep.saveStepAttribute(id_transformation, id_step, databaseName,databaseName); rep.saveStepAttribute(id_transformation, id_st

50、ep, where,where); rep.saveStepAttribute(id_transformation, id_step, columns,columns); rep.saveStepAttribute(id_transformation, id_step, tableName,tableName); rep.saveStepAttribute(id_transformation, id_step, limit,limit);.catch(Exception e)thrownew DataIntegrationException(Unable to save step inform

51、ation to the repository for id_step=+id_step, e);public String getXML()StringBuffer retval = new StringBuffer();retval.append( +XMLHandler.addTagValue(username,username)+Const.CR); retval.append( +XMLHandler.addTagValue(password,password)+Const.CR); retval.append( +XMLHandler.addTagValue(server,server)+Const.CR);retval.append(+XMLHandler.addTagValue(databaseName,databaseName)+Const.CR); retval.append(+XMLH