保持登录。
单击提交则表示您同意developerWorks 的条款和条件。 查看条款和条件.
在您首次登录 developerWorks 时，会为您创建一份个人概要。您的个人概要中的信息（您的姓名、国家/地区，以及公司名称）是公开显示的，而且会随着您发布的任何内容一起显示，除非您选择隐藏您的公司名称。您可以随时更新您的 IBM 帐户。
所有提交的信息确保安全。
当您初次登录到 developerWorks 时，将会为您创建一份概要信息，您需要指定一个昵称。您的昵称将和您在 developerWorks 发布的内容显示在一起。昵称长度在 3 至 31 个字符之间。 您的昵称在 developerWorks 社区中必须是唯一的，并且出于隐私保护的原因，不能是您的电子邮件地址。
单击提交则表示您同意developerWorks
的条款和条件。 .
所有提交的信息确保安全。
developerWorks 社区:
我的概要信息
选择语言：
Infosphere Datastage 是一个功能强大的数据抽取、转换、装载工具，广泛用于信息整合项目之中。它还提供了几十个数据转换 stage 及上百个数据转换函数，可以满足我们数据转换的需求。同时，Infosphere Datastage 还提供了多种扩展方法，方便用户开发满足特定业务需求的处理功能，重用常见业务处理过程，将串行执行的处理程序在 Datastage 中并行运行，进一步提高性能。本文以示例介绍的方式介绍了 Infosphere Datastage 常见的几种扩展方法及区别，各种扩展方式的使用场景及具体配置方法，希望能够使读者对 Infosphere Datastage 各种扩展方式有一个比较全面的了解。
(), 销售技术支持工程师, IBM
作者目前在 IBM 软件部工作，主要负责开发商在数据库和数据仓库方面产品的技术支持和培训，在 IT 行业有丰富的工作经验。是IBM 认证的 DB2 UDB 高级技术专家。
免费下载：或者 下载更多的 ，并加入 ，参与在线交流。我们知道， Infosphere Datastage 是一个功能强大的数据抽取、转换、装载工具，广泛用于用户信息整合项目之中。它不仅提供了丰富的数据接口，可以连接种类广泛的大型机、数据库、ERP/CRM 等企业应用及外部信息资源，它还提供了几十个数据转换 stage 及上百个数据转换函数，可以满足我们数据转换的需求，其中，比较常用的 Stage 主要包括：用于数据源及目标的 stage：各种数据库的 connector、sequential file、data set用于数据合并的 stage: lookup、joins、Merge、Aggregator用于数据转换的 stage: transformer、Remove Duplicates辅助的 stage: Row Generator、Peek、Sort通常来说，阶段提供绝大多数企业数据整合应用所需的 80% 到 90% 的应用逻辑，同时，InfoSphere DataStage 也提供了 c++ 及 java 编程接口，允许我们开发、扩展满足特定需求的转换功能。从底层技术架构来看，Infosphere Datastage 提供了一个并行、可扩展的应用框架，如下图所示，该并行、可扩展的应用框架主要由四个部分组成：应用组件、应用框架、配置文件及应用服务。每一个应用组件完成数据处理及应用分析的不同功能。其中：图 1. Datastage 并行扩展应用框架图 1. Datastage 并行扩展应用框架应用组件：是实现应用处理逻辑的基本处理单元，包括同数据库的接口、数据连接操作、分析、排序、分组聚合操作等，所有这些应用组件连接在一起来完成数据处理应用操作。该并行、 可扩展的应用框架还允许用户通过包装已有应用、开发 c/c++ 应用程序来定制自己的应用处理组件，所有这些组件均可以被复用，并且提供并行处理能力。通常来说，阶段提供绝大多数企业数据整合应用所需的 80% 到 90% 的应用逻辑，同时，InfoSphere DataStage 也提供了 c++ 及 java 编程接口，允许我们开发、扩展满足特定需求的转换功能。应用框架：用来构建并行处理应用。用户可以通过图形界面或命令行方式，选择各种处理组件，并将这些组件顺序连接在一起来实现各种处理流程。在内部，应用框架会自动实现作业的并行处理操作，包括负载均衡、分区间通讯、数据模式处理、多线程操作及数据缓冲操作等。用户不需要关心并行处理编程方法来实现并行处理效率，这些完全由应用框架自动完成。配置文件：用来指定数据处理应用在不同硬件平台上的并行处理能力。一般情况下，用户自己开发的应用程序，其应用逻辑及硬件平台编码往往交织在一起，因此，很难在不同硬件平台之间迁移。Infosphere Datastage 通过配置文件来隔离硬件平台编码，因此，可以通过改变配置文件的设置来轻松实现不同平台间应用迁移，而不需要改变应用逻辑。应用服务：用来提供数据处理应用的运行时态管理，包括跨越多个 cpu 处理器作业的单一程序及单一操作符的控制，集中错误信息的收集，性能监控，作业检查点重启控制等。Infosphere Datastage 中所提供的各种 stage，在其并行、可扩展的应用框架中都被影射成各种操作符（Operator），其中常见的一些 stage 所对应的操作符如下：表 1. Datastage 操作符对照表Infosphere Datastage
并行、可扩展的应用框架，通过数据分区、流水线及数据重分区等技术提供了高效的并行处理能力，充分保证数据整合的执行效率。所谓数据分区技术，就是通过哈希散列算法、数据轮训操作、随机操作等方式将数据分布到不同的数据分区，不同分区之间并行操作来提高数据处理效率。我们可以根据一台机器 CPU 的数据量及不同机器来设置数据分区的数量，比如，如果系统有 4 个 CPU，我们就可以将数据分布到 4 个数据分区上，理论上讲， 执行效率是单分区的 4 倍；所谓流水线技术，就像生产流水线一样，上一阶段数据处理完成后，会通过内存立刻传送到下一阶段进行处理，数据不落地，不产生 I/O 操作，所以效率会更高，同时，它结合数据分区技术，可以保证在每一个数据分区之间不落地进行数据处理，实现高效、并行处理。所谓数据重分区技术，是指可以根据数据处理的要求，重定义数据分区，尽量保证数据均匀分布到不同的数据分区，实现所谓 co-location 操作，即数据连接等操作尽可能在同一个数据分区内完成，这样执行效率最高。Infosphere Datastage 中所提供的所有数据处理 stage 都基于该并行、可扩展的应用框架进行开发的，所以，所有数据处理 stage 都提供并行处理能力。另外，Infosphere Datastage
并行、可扩展的应用框架还提供了非常灵活的可扩展能力，我们可以通过多种方式非常方便地开发自己特定的处理功能。在 Infosphere Datastage 中，扩展方法主要包括以下几种方法：包装阶段 (Wrapper stages)
通过包装已存在的应用程序来构建新的定制阶段 (stage) 来实现扩展功能构建阶段 (Build stages)
通过使用 DataStage 提供的图形界面创建新的阶段，并编译成为并行、可扩展的应用框架中新的操作符 (operator)。在定义新的阶段时，需要指定新阶段的属性、输入及输出接口、C++ 源代码。DataStage 会自动编译、链接 C++ 源代码，并生成新的操作符。定制阶段 (Custom stages)
通过使用 Infosphere Datastage 并行、可扩展应用框架提供的一个完整的 C++ API 开发复杂的、可扩展的阶段。使用该方法生成的新操作符均是基于 APT_Operator 类，C++ 源代码的编译及链接需要用户自己来完成。外部函数
用来定义在 Transformer 阶段中使用的新的并行处理例程。我们需要指定输入参数。C++ 函数的创建、编译及链接，需要用户自己来完成。我们通过下边的表格，简要地描述了上述几种扩展方式的优点及使用场景，供我们在进行 Datastage 功能扩展时选择。表 2. Datastage 扩展方法场景选择接下来，我们会介绍每种扩展方法的具体使用情况，希望可以使大家对 Datastage 扩展方法有一个比较全面、具体的了解。关于定制阶段扩展方法，需要对 Datastage 并行、可扩展的应用框架 API 有一个较为全面、深入的了解，限于篇幅，大家可以参考附录中 developerWorks 上的相关文章，这里就不在赘述了。包装阶段包装阶段可以将我们已有的 Unix shell 脚本、命令行程序、COBOL、 C、 C++、Java 应用程序及第三方应用程序构建成 Datastage 中的新的定制阶段 (stage) 来实现扩展功能。采用包装阶段扩展方法，还可以将原来串行运行的程序在 Datastage 中并行运行，进一步提高运行效率。如下图所示，对 Datastage 来说，包装阶段就是一个黑盒子，Datastage 并不了解包装阶段中的内容是什么，没有办法管理发生在包装阶段中的任何事情，它只了解如何数据如何输入到包装阶段以及数据如何从包装阶段输出，用户自己需要在设计阶段了解包装阶段需要执行哪些操作以及数据接口模式是什么。如果包装阶段应用需要在处理数据之前看到所有数据，该包装阶段应用不能并行执行。图 2. Datastage 包装阶段示意图图 2. Datastage 包装阶段示意图通过使用包装阶段扩展方法，我们可以得到如下好处：在不同的数据处理作业中重用处理逻辑。同串行执行相比，通过 datastage 应用框架提供的并行处理能力实现更好的可扩展行及处理性能。存在一些例外情况，比如某些应用可能不能并行执行，或者某些应用需要整个数据集保存在单个分区上，限制了并行处理能力。通过将复杂的处理逻辑包装成为包装阶段，可以避免在不同作业之间重复编写上述复杂处理逻辑。创建包装阶段的方法如下图所示，我们使用 Datastage Designer，在 Repository 窗格中，选择 Stage Type 文件夹，点击右键选择 New & Other & Parallel Stage Type(Wrapped) 来创建包装阶段。图 3. 创建包装阶段本次示例，我们将创建一个自定义的过滤阶段，用来执行一个简单的 Unix ‘ grep – s Straing ’命令。该自定义过滤阶段允许用户指定需要过滤的字符串，其执行结果将只输出包含指定字符串的记录。本次示例中，我们将定义如下数据模式作为自定义的过滤阶段 Improved_Grep 输入及输出数据项，如下边所示：清单 1. 过滤阶段 Improved_Grep 输入及输出数据模式清单 1. 过滤阶段 Improved_Grep 输入及输出数据模式 record
first_name:string[max=16] {cycle={
value='John',
value='Susan',
value='William',
value='Ann',
value='Frank',
value='Jane',
value='Seymour',
value='Laura'}};
last_name:string[max=20] {cycle={
value='Parker',
value='Calvin',
value='Mandella',
value='Claybourne',
value='Chalmers',
value='Studdock',
value='Glass',
value='Engels',
value='Boone',
value='Sarandon',
value='Tell',
value='Dillard',
value='Sinatra',
value='Austin',
value='Smith'}};
gender:string[1] {cycle={value='M', value='F'}};
birth_date:date {random={seed='part',
limit=10000},
epoch=''};
income:decimal[9,2] {random={seed='part',
limit=99999.99}};
state:string[2] {cycle={
value='MA',
value='IL',
value='CA',
value='FL',
value='NY',
value='TX',
value='NJ',
value='KY',
value='CO',
value='ND',
value='SD',
value='MI',
value='OH'}};
partition_number:uint8 {cycle={init='part',
partition_count:uint8 {cycle={init='partcount',
record_in_partition:uint32 {cycle={init=0,
record_id:uint32 {cycle={init='part',
incr='partcount'}};
)同时，我们将对 state 字段进行过滤，比如，我们需要只输出 state= ’ NY ’ 的所有记录。在创建包装阶段之前，我们需要将上述数据模式导入到 Datastage 元数据存储库中，如下图所示，我们选择 Import & Table Definitions &Orchestrate Schema Definition 导入数据模式定义到 Datastage 元数据库 Table Definitions& Orchestrate &SchemaImports&lab2_schema 中。图 4. 导入数据模式定义创建包装阶段 - 第一步图 5. 创建 Improved_Grep 包装阶段如上图所示，我们创建一个 Improved_Grep 包装阶段，在 Stage Type Name 栏目中输入 Improved_Grep，该名称将显示在 Datastage Designer 面板上；在 Wrapper Name 栏目中输入 Improved_Grep，用来指定包装阶段的名称；在 Execution Mode 栏目中，选择默认选项 Parallel Only，指定该阶段以并行方式执行，我们也可以选择串行执行方式；在 Command 栏目中输入我们需要执行的 grep 命令。创建包装阶段 - 第二步接下来，我们需要定义输入及输出接口。我们首先定义输入接口。如下图所示，我们选择 Wrapped 标签，并选择 Interfaces & Input 来定义输入接口。图 6. 定义 Improved_Grep 输入接口输入接口用来告诉 Datastage 将以什么样的数据格式将数据输入给包装阶段来对数据进行处理。Link 数据项用来定义包装阶段数据输入流的个数及每一个数据流的定义格式。我们可以定义多个输入流，Link 数据项从 0 开始。 本次示例，我们只定义了一个输入流。Table Name 数据项用来定义该数据流数据格式。我们需要事先定义好数据格式，本次示例中，我们将使用在上面步骤定义好的 Table Definitions& Orchestrate &SchemaImports&lab2_schema 数据定义。Stream 数据项用来定义数据流的属性。我们选择‘ use Stream ‘，并在 FileDescriptor 项中选择 Standard Input，在 Unix 环境下，我们通常选择该种方式。同时，它也支持命名管道方式。接下来我们定义输出接口。如下图所示，我们选择 Wrapped 标签，并选择 Interfaces & Output 来定义输出接口。图 7. 定义 Improved_Grep 输出接口输出接口用来告诉 Datastage 经包装阶段处理后的数据输出结果格式是什么。Link 数据项用来定义包装阶段数据输出流的个数及每一个数据流的定义格式。我们可以定义多个输出流，Link 数据项从 0 开始。 本次示例，我们只定义了一个输出流。Table Name 数据项用来定义该数据流数据格式。我们需要事先定义好数据格式，本次示例中，我们将使
用在上面步骤定义好的 Table Definitions& Orchestrate &SchemaImports&lab2_schema 数据定义。Stream 数据项用来定义数据流的属性。我们选择‘ use Stream ‘，并在 FileDescriptor 项中选择 Standard Output，在 Unix 环境下，我们通常选择该种方式。同时，它也支持命名管道方式。
创建包装阶段 - 第三步接下来，我们需要定义包装阶段运行时需要的环境变量及运行结果返回码。如下图所示，我们选择 Wrapped 标签，并选择 Environment 来定义环境变量及运行结果返回码。比如，我们可以定义当包装阶段运行成功后，返回结果码 1。在本次示例中，为了简便，我们选择了‘ All Exit Codes Successful ‘选项。图 8. 定义 Improved_Grep 环境变量及运行结果返回码创建包装阶段 - 第四步接下来，我们定义包装阶段所需要的输入参数信息。这一步不是必须的，只有包装阶段中定义了输入参数或选项，我们才需要定义该选项。如下图所示，我们选择 Properties 标签来定义包装阶段属性信息。图 9. 定义 Improved_Grep 输入参数信息本次示例中，我们定义了 Property Name= ‘ s ’ ，其显示属性 Prompt= ’ state number ’ 用来过滤不符合 state number 输入值的所有记录。State number 是州的缩写，因此，其数据类型定义为 string，Conversion 数据项我们选择 -Name Value，其默认值我们选择了 MA，当不输入参数时，我们将只选择州缩写为 MA 的记录。创建包装阶段 - 第五步最后，我们重新选择 General 标签，并依次选择 Generate 按钮及 OK 按钮来最终创建完成包装阶段。如下图所示：图 10. 创建完成 Improved_Grep我们创建的 Improved_Grep 包装阶段将保存在 Datastage 元数据存储库的 Rountines 目录下，并在 Datastage Desiner 工具面板的 Runtines 目录中显示，如下图所示：图 11. Improved_Grep 元数据存储库系统为 Improved_Grep 包装阶段生成默认图标，如下图所示，我们也可以定制自己的显示图标，我们可以选择 Creator 标签，并点击 16X16 Bitmap 及 32X32 Bitmap 图标为 Improved_Grep 包装阶段选择自己的大、小图标显示。图 12. 改变 Improved_Grep 图标使用 Improved_Grep 包装阶段接下来，我们创建一个简单的 Datastage 作业来使用我们刚刚创建的 Improved_Grep 包装阶段来选择 state 为 NY 的所有记录。如下图所示：图 13. 使用 Improved_Grep 包装阶段的作业我们使用 Row_Generator 阶段按照以下数据模式生成 1000 条记录：清单 2. Row_Generator 阶段使用的数据模式清单 2. Row_Generator 阶段使用的数据模式 record
first_name:string[max=16] {cycle={
value='John',
value='Susan',
value='William',
value='Ann',
value='Frank',
value='Jane',
value='Seymour',
value='Laura'}};
last_name:string[max=20] {cycle={
value='Parker',
value='Calvin',
value='Mandella',
value='Claybourne',
value='Chalmers',
value='Studdock',
value='Glass',
value='Engels',
value='Boone',
value='Sarandon',
value='Tell',
value='Dillard',
value='Sinatra',
value='Austin',
value='Smith'}};
gender:string[1] {cycle={value='M', value='F'}};
birth_date:date {random={seed='part',
limit=10000},
epoch=''};
income:decimal[9,2] {random={seed='part',
limit=99999.99}};
state:string[2] {cycle={
value='MA',
value='IL',
value='CA',
value='FL',
value='NY',
value='TX',
value='NJ',
value='KY',
value='CO',
value='ND',
value='SD',
value='MI',
value='OH'}};
partition_number:uint8 {cycle={init='part',
partition_count:uint8 {cycle={init='partcount',
record_in_partition:uint32 {cycle={init=0,
record_id:uint32 {cycle={init='part',
incr='partcount'}};
)使用 Improved_Grep 包装阶段，并在属性栏中指定 state number=NY 来选择 state number=NY 的所有数据，如下图所示，并将结果输出到一个文本文件中。查看其输出结果，将只能看到 state number=NY 的记录。图 14. Improved_Grep 包装阶段属性信息构建阶段构建阶段扩展方式，相当于一个操作符构建器，用来将标准的 C++ 语句创建为 Datastage 并行扩展应用框架中的操作符，完成自己特定的数据处理功能，而且不需要对 C++ 有十分深入的理解。正如我们上边讲述 Datastage 并行扩展应用框架时所提到的，在 datastage 中的阶段，在运行时会转换成 Datastage 并行扩展应用框架的操作符，完成不同数据处理操作。1 个 datastage 阶段，可能对应多个操作符，如顺序文件阶段会影射成 Import 或 Export 操作符。在使用构建阶段扩展方式时，不需要对 C++ 有十分深入的了解，我们只需要在 Per-Record 部分输入自己的核心应用逻辑即可。举个例子，我们需要编写一段简单的 C++ 代码，实现汇总 2 个字段值，并输出到 sum_f1f2 字段，通常的 C++ 代码如下：清单 3. 求和计算 C++ 代码 #include &stdio.h&
int main()
int field1;
int field2;
int sum_f1f2;
sum_f1f2 = field1 + field2;
}在使用构建阶段扩展方式时，我们不需要编写完整的 C++ 代码，只需要在 Per-Record 部分输入下边一行代码即可：清单 4. 在 Per-Record 中 C++ 代码 um_f1f2 = field1 + field2;Datastage 提供的 buildop 实用程序会自动生成相应的 C++ 源程序，编译并链接生成动态库。所以，一般我们只需要了解 C++ 语言的编程语法、变量定义及声明、条件处理语句即可，不需要成为 c 或 C++ 语言的专家，在背后，Datastage 帮助我们完成 C++ 程序的编写、编译及链接操作。因此，我们需要系统安装 C++ 编译器。通过使用构建阶段扩展方法，我们可以得到如下好处：提供更好的灵活性
通过创建构建阶段，我们可以以并行处理方式来运行自定义的数据处理逻辑；并且可以将原来由多个复杂的阶段实现的处理逻辑用一个构建阶段来实现，提高代码的复用性。提供更快的处理速度
构建阶段以本机方式在 Datastage 并行扩展应用框架中运行，和 Datastage 内置提供的阶段运行效率是一样的；自己编写的代码也往往比 Transformer 阶段自动生成的代码效率更高。提供更好的可移植性
代码是使用标准的 ANSI C/C++ 编写，具有更好的可移植性；同时，该阶段可以在任何地方被复用。创建构建阶段的方法如下图所示，我们使用 Datastage Designer，在 Repository 窗格中，选择 Stage Type 文件夹，点击右键选择 New & Other & Parallel Stage Type(Build) 来创建构建阶段。图 15. 创建 TotalItemsAmount 构建阶段本次示例，我们将创建一个自定义的计算阶段 TotalItemsAmount，用来计算产品的销售收入。TotalItemsAmount 计算阶段的输入数据包括 Qty、Price、Tax_Rate 三个字段，输出数据包括 Amount 字段。另外，我们还需要定义一个 Exchange 属性字段，用来计算汇率转换之后的销售额。因此，我们需要执行下边的逻辑来计算产品的销售收入：清单 5. 计算产品的销售收入 beforeTaxAmount = InRec.Qty *P
tax=Tax_Rate*beforeTaxA
Amount=beforeTaxAmount+
Amount=Amount*E在定义 TotalItemsAmount 计算阶段时，在开发过程中，我们有时需要加入一些调试功能，以便更容易排错。在本次示例中，我们定义了一个 Debug 变量，包含 0，1，2 三个值，默认值为 0。 当我们选择 Debug 变量值为 1 或 2 时，TotalItemsAmount 计算阶段将在作业开始时在 Datastage 日志文件中输出“Debugging level n has been enabled”信息；当 Debug 变量值为 1 时，TotalItemsAmount 计算阶段同时在 Datastage 日志文件中记录每一行记录的输入、输出值；当 Debug 变量值为 2 时，TotalItemsAmount 计算阶段还要在 Datastage 日志文件中记录 beforeTaxAmount 及 tax 中间变量的值；最后，当我们选择 Debug 变量值为 1 或 2 时，TotalItemsAmount 计算阶段还将在读入所有记录后在 Datastage 日志文件中输出“All rows have been processed by the BuildOp”信息。另外，我们还需要检查 Qty 字段输入值，当 InRec.Qty&=0 ||InRec.Qty&750 ，我们需要在 Datastage 日志文件中记载相应出错信息。我们为 TotalItemsAmount 计算阶段的输入接口定义了 custin 表定义模式，并将其保存到 Datastage 元数据存储库 Table Definitions&custin 中，如下边所示：清单 6. custin 表定义模式 record
{final_delim=end, delim=',', quote=double}
Qty:int32;
)我们为 TotalItemsAmount 计算阶段的输出接口定义了 custout 表定义模式，并将其保存到 Datastage 元数据存储库 Table Definitions&custout 中，如下边所示：清单 7. custout 表定义模式 record
{final_delim=end, delim=',', quote=double}
)关于输入数据中的其他字段，我们选择 RCP 方式传递给 TotalItemsAmount 计算阶段。创建构建阶段 - 第一步图 16. TotalItemsAmount 构建阶段总体信息如上图所示，我们创建一个 TotalItemsAmount 构建阶段，在 Stage Type Name 栏目中输入 TotalItemsAmount，该名称将显示在 Datastage Designer 面板上；在 Class Name 栏目中输入 TotalItemsAmount ，将创建 TotalItemsAmount 类；在 Execution Mode 栏目中，选择默认选项 Parallel Only，指定该阶段以并行方式执行，我们也可以选择串行执行方式；在 Operator 栏目中输入 TotalItemsAmount，指定 Datastage 并行扩展应用框架中引用的操作符名称，通常，该名称同阶段名称保持一致。创建构建阶段 - 第二步接下来，我们需要定义输入及输出接口。我们首先定义输入接口。如下图所示，我们选择 Build 标签，并选择 Interfaces & Input 来定义输入接口。图 17. TotalItemsAmount 输入接口定义输入接口用来告诉 Datastage 将以什么样的数据格式将数据输入给构建阶段来对数据进行处理。我们只需要指定构建阶段使用的字段，所有在输入接口定义的字段在构建阶段代码中必须使用。Port Name 数据项：当构建阶段存在多个输入或输出流时，我们需要为每一个输入或输出流创建接口，并需要定义 Port Name 来区分不同的输入或输出流。当构建阶段只有一个输入及输出流时，我们可以不定义 Port Name。本次示例中 , 我们为输入流定义了 InRec 端口名称；Table Name 数据项：用来定义该数据流数据格式，我们只需要定义构建阶段需要的字段即可。由于构建阶段采用静态接口方式，因此，输入及输出接口数据模式定义要事先定义，本次示例中，我们将使用在上面步骤定义好的 Table Definitions&custin 数据定义。Auto Read 数据项：默认值为 True，表明该阶段自动通过端口读入数据。如果该选项设置为 False，我们需要在代码中自己控制数据读入方式。本次示例中 , 我们设置为 True。RCP 数据项：默认值为 False，当设置为 True，将强制将没有在输入接口中定义的字段传递给下流处理阶段。如果设置为 False，没有在输出接口中定义的字段不会传递给下流处理阶段。本次示例中 , 我们设置为 True。接下来我们定义输出接口。如下图所示，我们选择 Build 标签，并选择 Interfaces & Output 来定义输出接口。图 18. TotalItemsAmount 输出接口定义输出接口用来告诉 Datastage 经构建阶段处理后的数据输出结果格式是什么。我们只需要指定构建阶段使用的字段，所有在输出接口定义的字段在构建阶段代码中必须使用。Port Name 数据项：当构建阶段存在多个输入或输出流时，我们需要为每一个输入或输出流创建接口，并需要定义 Port Name 来区分不同的输入或输出流。当构建阶段只有一个输入及输出流时，我们可以不定义 Port Name。本次示例中 , 我们为输出流定义了 OutRec 端口名称；Table Name 数据项：用来定义该数据流数据格式，我们只需要定义构建阶段需要的字段即可。由于构建阶段采用静态接口方式，因此，输入及输出接口数据模式定义要事先定义，本次示例中，我们将使用在上面步骤定义好的 Table Definitions&custout 数据定义。Auto Write 数据项：默认值为 True，表明该阶段自动通过端口读出数据。一旦数据输出后，该数据就不能在构建阶段中再次使用。如果该选项设置为 False，我们需要在代码中自己控制数据读出方式。本次示例中 , 我们设置为 True。RCP 数据项：默认值为 False，当设置为 True，将强制将没有在输入接口中定义的字段传递给下流处理阶段。如果设置为 False，没有在输出接口中定义的字段不会传递给下流处理阶段。本次示例中 , 我们设置为 True。下边，我们来定义传输接口。如下图所示，我们选择 Build 标签，并选择 Interfaces & Transfer 来定义传输接口。图 19. TotalItemsAmount 传输接口定义传输接口允许用户定义数据在输入接口及输出接口之间传输的方式。一般情况下，如果存在多个输入或输出流时需要定义该接口。Auto Transfer 数据项：默认值为 False，表明你需要自己编码来控制数据传输方法。如果设置为 True，构造阶段自动处理数据传输方法。Separate 数据项：默认值为 False，表明后边数据流将覆盖前边数据流相同字段的数据，相当于 Remove duplicates 操作。如果设置为 True，后边的数据流将合并在前边数据流后边，如果前边数据流最后字段同后边数据流前面字段相同，将去除相同的字段。本次示例中 , 只有一个输入及输出流，因此，我们没有定义该接口。创建构建阶段 - 第三步接下来，我们需要定义 TotalItemsAmount 构建阶段属性信息。如下图所示，我们选择 Properties 标签来定义 TotalItemsAmount 构建阶段属性信息。图 20. TotalItemsAmount 属性信息定义本次示例中，我们定义了 Exchange 属性变量，用来表示汇率信息，其类型为 Float，默认值为 1，Conversion 数据项我们选择 -Name Value；我们定义了 Debug 属性变量，用来指定调试级别，其类型为 String，默认值为 0，Conversion 数据项我们选择 -Name Value。Exchange 及 Debug 属性变量，我们都会在逻辑处理代码中使用。创建构建阶段 - 第四步接下来，我们需要定义 TotalItemsAmount 构建阶段核心处理逻辑所需要的变量。如下图所示，我们选择 Build 标签，并选择 Logic &Definitions 来定义 TotalItemsAmount 构建阶段核心处理逻辑所需要的变量。图 21. TotalItemsAmount 核心处理逻辑变量定义我们可以在此定义需要的变量并引用头文件信息。变量的类型可以是标准 C 的类型，也可以是 Datastage 并行扩展应用框架内置类型。Datastage 并行扩展应用框架内置类型主要包括如下类型：APT_String、APT_Int32、APT_Date、APT_Decimal、APT_TimeStamp。这些类型比标准 C 类型具有更高级的处理能力，因此，建议尽量使用这些类型。本次示例中 , 我们定义了 beforeTaxAmount 及 tax 变量。创建构建阶段 - 第五步接下来，我们需要定义 TotalItemsAmount 构建阶段 Pre-Loop 或 Post-Loop 处理逻辑。如下图所示，我们选择 Build 标签，并选择 Logic &Pre-Loop 或 Post-Loop 来定义 TotalItemsAmount 构建阶段 Pre-Loop 或 Post-Loop 处理逻辑。图 22. TotalItemsAmount 阶段 Pre-Loop 或 Post-Loop 处理逻辑Pre-Loop 标签：用来定义当所有记录被处理之前执行的处理逻辑。我们可以使用 C/C++ 语言编写这些处理逻辑。Post-Loop 标签：用来定义当所有记录被处理之后执行的处理逻辑。我们可以使用 C/C++ 语言编写这些处理逻辑。本次示例中 , 我们创建了如下 Pre-Loop 处理逻辑来实现：当我们选择 Debug 变量值为 1 或 2 时，TotalItemsAmount 计算阶段将在作业开始时在 Datastage 日志文件中输出“Debugging level n has been enabled”信息。清单 8.
Pre-Loop 处理逻辑 if (Debug=="1"||Debug=="2") {
cout&& "Debug lebel" && Debug &&"has been enabled"&&
}另外，我们创建了如下 Post-Loop 处理逻辑来实现：当我们选择 Debug 变量值为 1 或 2 时，TotalItemsAmount 计算阶段将在读入所有记录后在 Datastage 日志文件中输出“All rows have been processed by the BuildOp”信息。清单 9.
Post-Loop 处理逻辑 if (Debug=="1"||Debug=="2") {
cout && "All rows have been processed by the BuildOp" &&
}创建构建阶段 - 第六步接下来，我们需要定义 TotalItemsAmount 构建阶段的核心处理逻辑。如下图所示，我们选择 Build 标签，并选择 Logic &Per- Record 来定义 TotalItemsAmount 构建阶段的核心处理逻辑。图 23. TotalItemsAmount 阶段 Per-Record 定义我们需要在 Per-Record 标签中输入核心处理逻辑。我们可以编写任何 ANSI C/C++ 代码，同时可以引用 Datastage 并行扩展应用框架内置的函数及宏。在 Per-Record 标签中定义的处理逻辑，针对每一行记录进行处理，当记录输出后，不能再被使用。我们还可以定义记录的缓冲机制，以及管理记录的输入、输出流程。在定义处理逻辑时，我们可以直接使用字段，不需要事先声明。本次示例中 , 我们在 Per-Record 标签中主要实现了三部分处理逻辑：一部分用来计算销售额 Amount，另一部分用来检查 Qty 输入数据是否在 InRec.Qty&=0 ||InRec.Qty&750 范围内，最后一部分是通过检查 Debug 变量的设置，来输出不同的日志及调试信息。详细代码信息如下：清单 10.
Per-Record 处理逻辑 if (InRec.Qty&=0 ||InRec.Qty&750) {
*errorLog()&&"Qty outside of range. Value="&&InRec.Qty&&
errorLog().logWarning(0);
if (Debug=="1" ||Debug =="2") {
cout&& "Qty="&&InRec.Qty&&
cout&& "Price="&&InRec.Price&&
cout&& "Tax_Rate="&&InRec.Tax_Rate&&
beforeTaxAmount=InRec.Qty *P
if (Debug=="2") {
cout&&"beforeTaxAmount="&&beforeTaxAmount&&
tax=Tax_Rate*beforeTaxA
if (Debug=="2") {
cout&&"tax="&&tax&&
Amount=beforeTaxAmount+
Amount=Amount*E
if (Debug=="1"||Debug=="2") {
cout&&"Amount="&&Amount&&
}在 Datastage 中，我们需要使用 errorLog 对象来输出错误信息到 Datastage 日志文件中，默认情况下，警告信息用黄色标记表示，错误信息用红色标记来表示。errorLog 对象定义在 errorlog.h 头文件中。我们可以使用 *errorLog() && “Message to write” && endl 来输出错误信息到 Datastage 日志文件；errorLog().logInfo(index) 来生成日志信息；errorLog().logWarning(index) 来生成警告信息；errorLog().logError(index) 来生成错误信息。本次示例中，我们使用如下语句来将"Qty outside of range. Value="信息写入 Datastage 日志文件中，并以警告信息方式记录。清单 11.
记录警告日志信息 *errorLog()&&"Qty outside of range. Value="&&InRec.Qty&&
errorLog().logWarning(0);另外，Datastage 还提供了丰富的宏，我们可以在代码中直接使用。下边列出了 Datastage 中比较常见的宏 :表 3. Datastage 提供的常用宏信息创建构建阶段 - 第七步最后，我们点击 Generate 按钮编译程序代码，最终生成 TotalItemsAmount 构建阶段。如下图所示：图 24. 生成 TotalItemsAmount 构建阶段当出现“Operator Generation Succeeded”，表明我们成功创建了 TotalItemsAmount 构建阶段。我们创建的 TotalItemsAmount 构建阶段将保存在 Datastage 元数据存储库的 Rountines 目录下，并在 Datastage Desiner 工具面板的 Runtines 目录中显示，如下图所示：图 25. TotalItemsAmount 元数据存储库系统为 TotalItemsAmount 构建阶段生成默认图标，如下图所示，我们也可以定制自己的显示图标，我们可以选择 Creator 标签，并点击 16X16 Bitmap 及 32X32 Bitmap 图标为 TotalItemsAmount 构建阶段选择自己的大、小图标显示。图 26. 改变 TotalItemsAmount 图标当我们使用 Datastage Desiner 构建 TotalItemsAmount 构建阶段时，Datastage 会创建 TotalItemsAmount.opd 文件，Datastage 的 buildop 实用程序会根据 TotalItemsAmount.opd 文件自动生成 TotalItemsAmount.C 源代码，编译并链接生成 TotalItemsAmount.so 动态库文件。
系统自动生成的源文件、动态库文件会保存在 /opt/IBM/InformationServer/Server/Projects/dstage1/buildop 目录下，具体文件包括：
buildop 目录下文件信息 -rw-rw-rw- 1 dsadm dstage
18:46 operator.apt
-rw-rw-rw- 1 dsadm dstage
18:46 operator.apt.bak
-rw-rw-r-- 1 dsadm dstage -05-08 18:45 TotalItemsAmount.C
-rw-rw-r-- 1 dsadm dstage
-08 18:45 TotalItemsAmount.h
-rw-rw-r-- 1 dsadm dstage
-08 18:45 TotalItemsAmount.opd
-rw-rw-r-- 1 dsadm dstage
18:45 TotalItemsAmount.opd.sh
-rwxrwxr-x 1 dsadm dstage -05-08 18:46 TotalItemsAmount.so其中，TotalItemsAmount.opd 文件包含了 TotalItemsAmount 构建阶段所需要的所有信息：清单 13.
TotalItemsAmount.opd 文件信息 $operator TotalItemsAmount
$class TotalItemsAmount
$input InRec auto record( Qty:int32; Price: Tax_Rate: inRec:*;)
$output OutRec auto record( Amount: outRec:*;)
$argument Exchange float
$argument Debug string
float beforeTaxA
if (Debug=="1"||Debug=="2") {
cout&& "Debug lebel" && Debug &&"has been enabled"&&
if (InRec.Qty&=0 ||InRec.Qty&750) {
*errorLog()&&"Qty outside of range. Value="&&InRec.Qty&&
errorLog().logWarning(0);
if (Debug=="1" ||Debug =="2") {
cout&& "Qty="&&InRec.Qty&&
cout&& "Price="&&InRec.Price&&
cout&& "Tax_Rate="&&InRec.Tax_Rate&&
beforeTaxAmount=InRec.Qty *P
if (Debug=="2") {
cout&&"beforeTaxAmount="&&beforeTaxAmount&&
tax=Tax_Rate*beforeTaxA
if (Debug=="2") {
cout&&"tax="&&tax&&
Amount=beforeTaxAmount+
Amount=Amount*E
if (Debug=="1"||Debug=="2") {
cout&&"Amount="&&Amount&&
if (Debug=="1"||Debug=="2") {
cout && "All rows have been processed by the BuildOp" &&
$end当用户在 Datastage Designer 界面中点击 Generate 按键生成 TotalItemsAmount 构建阶段时，Datastage 会调用 TotalItemsAmount.opd.sh 命令根据 TotalItemsAmount.opd 描述的信息来生成 TotalItemsAmount.C，TotalItemsAmount.h 文件，并编译、链接生成 TotalItemsAmount.so 文件。使用 TotalItemsAmount 构建阶段接下来，我们创建一个简单的 Datastage 作业，从 order_items.txt 文件中读取订单销售数据，使用我们刚刚创建的 TotalItemsAmount 构建阶段计算销售额，并将计算结果输出到一个文件中用于后续操作。如下图所示：图 27. 包含 TotalItemsAmount 的作业我们使用 Sequential File 阶段读取 order_items.txt 文件中的订单数据，其数据格式包括 OrderID, ItemNumber, Quantity, Price, TaxRate5 个字段。由于 order_items.txt 中的定义的 Quantity, TaxRate 字段，在 TotalItemsAmount 构建阶段输入接口中对应的是 Qty, Tax_Rate，因此，我们使用 Copy 阶段来进行字段名称的转换，如下图所示：图 28. 使用 Copy 阶段来进行字段名称的转换使用 TotalItemsAmount 构建阶段，检查 Qty 输入数据是否在 InRec.Qty&=0 ||InRec.Qty&750 范围内，计算销售额 Amount，并通过检查 Debug 变量的设置，来输出不同的日志及调试信息。如下图所示：图 29. 设置 TotalItemsAmount 构建阶段属性在属性栏中指定 Debug 及 Exchange 属性变量值，将计算结果输出到一个文本文件中，并将日志及调试信息输出到 Datastage 日志文件中。本次示例中，我们指定 Debug=2，Datastage 日志文件中相关日志及调试信息如下图所示：图 30. Datastage 日志输出信息外部函数在 Infosphere Datastage 中，我们不仅可以通过创建自定义阶段来扩展数据处理功能，我们还可以通过创建并行处理过程来扩展数据处理功能。Infosphere Datastage 的并行处理过程主要包括以下两种类型：外部函数
我们可以创建自定义的外部函数，用于 Transformer 阶段的字段表达式定义以及约束条件设置来扩展数据处理功能。这是我们扩展 Transformer 阶段数据处理功能最常使用的方法。外部函数需要返回值。外部 Before/After 过程
主要用来定义 Datastage 作业执行之前需要做哪些预处理操作，以及 Datastage 作业执行后需要做哪些处理操作。也可以用来定义 Transformer 阶段处理数据之前需要做哪些预处理操作，以及 Transformer 阶段处理完数据之后需要做哪些处理操作。外部 Before/After 过程没有返回值。我们可以在作业属性定义或 Transformer 阶段属性定义中来指定外部 Before/After 过程，如下图所示：图 31. 在作业属性中定义外部 Before/After 过程图 32. 在 Transformer 中定义外部 Before/After 过程在创建并行处理过程，包括外部函数及外部 Before/After 过程时，我们需要自己编写 C/C++ 函数或过程代码，并编译、链接生成静态目标文件或动态库函数。接下来，我们需要使用 DataStage Designer 工具定义并行处理过程的元数据信息，包括输入、输出参数，静态或动态链接方式。创建外部函数的方法如下图所示，我们使用 Datastage Designer，在 Repository 窗格中，选择 Routines 文件夹，点击右键选择 New & Parallel Routine 来创建外部函数。图 33. 创建外部函数本次示例，我们将创建一个外部函数 keyWords，用来判断输入的字符串是否是关键字。我们指定字符串”hello”及”ugly”为关键字，当输入字符串为关键字，我们返回字符串”Yes”，否则返回”No”。创建外部函数 - 第一步我们需要创建 keyWords.C 程序来实现上述逻辑，具体代码如下所示：清单 14.
keyWords.C 程序代码 #include &string&
#include &iostream&
char * keyWords(char *inString)
if (strstr(inString, "hello") !=NULL ||
strstr(inString, "ugly") !=NULL)
return "Yes";
return "No";
}接下来，我们需要编译并链接 keyWords.C 程序生成 keyWords.o 目标文件。本次示例使用的是 suse Linux 操作系统，因此，我们采用如下命令来编译生成 keyWords.o 目标文件：清单 15.
编译生成 keyWords.o 目标文件 g++ -c -m32 -Wno-deprecated -fPIC -shared -m32 -c keyWords.C当执行完成上述命令后，会在当前目录下生成 keyWords.o 目标文件。创建外部函数 - 第二步接下来，我们需要在 Datastage Designer 中创建 keyWords 外部函数。如下图所示：图 34. 创建 keyWords 外部函数在 Routine name 栏目中输入 keyWords，该名称将显示在 Datastage Designer 面板上；在 Type 栏目中选择 External Function，我们本次示例将创建一个外部函数。我们也可以选择 External Before/After Routine 选项来创建 Before/After 过程；在 External subroutine name 栏目中输入 keyWords ，该名称指定外部程序名称；在 Object type 栏目中选择 Object，本次示例中，我们将引用静态目标文件。我们也可以选择 Library，用来引用动态链接库；在 Return type 栏目中选择 char*，本次示例中，我们将返回字符串类型。根据需要，我们也可以选择其他数据类型；在 Library path 栏目中，输入 /test/keyWords.o，该文件是我们在第一步生成的静态目标文件。创建外部函数 - 第三步接下来，我们需要定义输入参数。如下图所示，我们选择 Arguments 标签来定义输入参数。图 35. 定义输入参数我们需要在这里定义所有的输入参数。输出参数是在第二步中 General 标签页面定义的。本次示例中，我们只定义了一个输入参数 inString，其 I/O 类型为 I，表示为输入参数，系统自动输入 I，不能修改；其数据类型为 char*，字符串类型。创建外部函数 - 第四步最后，我们点击 Save 按钮最终生成 keyWords 外部函数。如下图所示：图 36. 生成 keyWords 外部函数我们创建的 keyWords 外部函数将保存在 Datastage 元数据存储库的 Rountines 目录下，并在 Datastage Desiner 工具面板的 Runtines 目录中显示，如下图所示：图 37. keyWords 外部函数元数据存储库使用 keyWords 外部函数接下来，我们创建一个简单的 Datastage 作业，来使用我们刚刚创建的 keyWords 外部函数来识别输入字符串是否为我们定义的关键字。如下图所示：图 38. 使用 keyWords 外部函数的抽取作业我们使用 Row_Generator 阶段生成 1000 条记录，记录结构包含 inString 一个字段，其数据结构为 varchar 类型，生成的数据中包含’ hello ’，’ ugly ’字符串数据。使用 Transformer 阶段， 在字段表达式定义栏目中引用 keyWords 外部函数来识别输入字符串是否为我们定义的关键字。如下图所示：图 39. 在 Transformer 阶段引用 keyWords 外部函数我们可以在 Routines 文件夹中找到我们创建的 keyWords 外部函数，在字段表达式定义栏目中我们定义了如下语句：清单 16.
引用 keyWords 外部函数表达式 If keyWords(DSLink3.inString) ="Yes" Then "Found" Else
"Not Found"最后，我们将 Transformer 阶段处理结果输出到一个文件中。查看其输出结果，将看到包含’ hello ’，’ ugly ’字符串数据的记录将显示"Found"，其他记录显示为"Not Found"。结论Infosphere Datastage 提供了多种灵活的扩展方式来增强用户数据转换、处理能力。同时， 还可以利用这种扩展能力，将我们现有的处理程序扩充到 Infosphere Datastage 并行处理框架中，充分利用我们已有的资源。本文中，重点介绍了包装阶段、构建阶段及外部函数扩展方法，我们还可以使用定制阶段扩展方法调用 Infosphere Datastage 并行扩展处理框架提供的 C++ API 来开发满足自己特定需求的处理逻辑。Infosphere Datastage 内置提供的各种阶段基本都是采用 Infosphere Datastage 并行扩展处理框架提供的 C++ API 开发出来的。有关定制阶段扩展方法的具体介绍，请参考 Information Server 信息中心的相关内容。
参考 ，了解更多关于地理信息系统的详情。参考 ，了解数据库激活空间特性所要具备的条件。参考 ，来获取关于空间参考系的知识。参考 ，获得关于 sdelayer 指令的详细用法。参考 ，了解 DB2 空间扩展。：观看面向初学者的产品安装和设置演示，以及为经验丰富的开发人员提供的高级功能。在 ，了解关于信息管理的更多信息，获取技术文档、how-to 文章、培训、下载、产品信息以及其他资源。随时关注 developerWorks 和 。加入 ，developerWorks 社区是一个面向全球 IT 专业人员，可以提供博客、书签、wiki、群组、联系、共享和协作等社区功能的专业社交网络社区。加入 ，参与在线交流。
developerWorks: 登录
标有星（*）号的字段是必填字段。
保持登录。
单击提交则表示您同意developerWorks 的条款和条件。 查看条款和条件。
在您首次登录 developerWorks 时，会为您创建一份个人概要。您的个人概要中的信息（您的姓名、国家/地区，以及公司名称）是公开显示的，而且会随着您发布的任何内容一起显示，除非您选择隐藏您的公司名称。您可以随时更新您的 IBM 帐户。
所有提交的信息确保安全。
选择您的昵称
当您初次登录到 developerWorks 时，将会为您创建一份概要信息，您需要指定一个昵称。您的昵称将和您在 developerWorks 发布的内容显示在一起。昵称长度在 3 至 31 个字符之间。
您的昵称在 developerWorks 社区中必须是唯一的，并且出于隐私保护的原因，不能是您的电子邮件地址。
标有星（*）号的字段是必填字段。
(昵称长度在 3 至 31 个字符之间)
单击提交则表示您同意developerWorks 的条款和条件。 .
所有提交的信息确保安全。
IBM PureSystems(TM) 系列解决方案是一个专家集成系统
通过学习路线图系统掌握软件开发技能
软件下载、试用版及云计算
static.content.url=/developerworks/js/artrating/SITE_ID=10Zone=Information ManagementArticleID=947932ArticleTitle=Infosphere Datastage 扩展方法应用实践publish-date=