摘要: 本文介绍了一些可用来为通过 DataStage 作业将数据从来源系统传送到目标系统提供保障的方法。由于所访问的资源具有不同的性质，一种解决方案可能无法完美地应对所有情形，所以需要结合使用多种技术。确定使用何种方 ...

简介

InfoSphere DataStage 提供了一些强大功能，从来源系统提取数据，转换这些数据，并将它加载到目标系统中。许多 DataStage 用户需要获得能够保障数据从来源系统转移到目标系统并且绝不会丢失数据的解决方案。本文将详细介绍您可以在 InfoSphere DataStage 中用于保障数据从来源到目标的传送的各种方法。文中将介绍如何使用 Distributed Transaction Stage (DTS) 和一个事务管理器来处理分布式 XA 事务，并且不使用事务管理器来处理本地数据库事务。文本还将介绍如何使用多个输入链接和一个数据库连接器在一个本地事务内执行多个数据库操作，以及如何在将数据从 InfoSphere Data Replication 移动到 DataStage 时，使用 Change Data Capture Transaction 阶段完成有保障的传送。最后，还将介绍结合使用消息极端和数据库阶段来提供可靠的数据处理和转换服务的最佳实践。

全局事务与 DTS

本节大体介绍了 XA 2 阶段提交架构，展示了 DTS 如何通过与 WebSphere? MQ 和数据库资源交互来完成全局事务。

X/Open Group 标准 eXtended Architecture (XA) 定义了一个协议，在单个事务内更新多个资源。资源可以是数据库，也可以是消息系统，比如 WebSphere MQ。通过不可靠的网络连接跨多种资源实现原子性、一致性、隔离性和耐久性 (ACID)，这是一个亟待解决的困难问题。XA 使用一种 2 阶段提交 (2PC) 协议来实现此目的。事务管理器管理该协议，与多个管理数据库或消息资源事务的资源管理器进行通信。2PC 协议的两个阶段是：

1、提交请求：在此阶段中，事务管理器将请求消息发送到每个资源。这些资源将事务准备到提交点，而不实际提交事务。例如，一个资源可将数据写入数据库，但将每个记录标记为未提交。然后，该资源向事务管理器发送回一条状态消息，表明是成功了还是失败了。

2、提交：如果所有资源在第一阶段中都被报告成功，那么资源管理器会向每个资源都发送一条提交消息。这些资源会完成事务，然后向资源管理器报告它们的最终状态。如果资源管理器从任何资源收到任何失败消息，它会向所有资源发送一条中止消息，该消息会回滚它们未提交的事务。

2PC 之所以有效，是因为如果一个资源在第一阶段报告成功，那么它就为它将要提交数据做出了保证。支撑该协议的是一组握手消息，这些消息可以处理各种不同的清理条件，比如网络故障或事务中涉及的任何组件的故障。

DTS 使用 WebSphere MQ 事务管理器（WebSphere MQ Server 中的一个标准功能）来协调事务与资源管理器，从而利用 2PC，如 图 1 中所示：

图 1. DTS、MQ 和资源管理器之间的关系

DTS 首先连接到 WebSphere MQ，告诉事务管理器启动 XA 事务。事务管理器与资源管理器进行通信，以启动事务。DTS 然后利用 DB2、Oracle 或 WebSphere MQ 连接器向资源执行写入操作。最后，DTS 告诉事务管理器提交或回滚 XA 事务。

DTS 作业通常使用 WebSphere MQ 作为数据源，使用数据库作为目标。通过使用 DTS，从来源队列到目标数据据库的传送就得到了保障：如果事务被成功提交，则会从来源队列中删除来源消息，数据会写入到目标数据库。如果该作业失败，则不会向目标数据库写入任何数据，该消息也会保留在来源队列上。这就提供了保障，因为来源消息的删除是在与目标数据库的更新相同的 XA 事务内执行的，因此两种操作要么都会执行，要么都不执行。

DTS 作业通常使用工作队列 将工作分区为并行管道。工作文件的使用最好借助图表来解释，如图 2 所示：

图 2. DTS 对工作队列的使用

这个作业与 WebSphere MQ 一起通过以下方式保障数据的传送：

1.在本地同步点控制下，MQ Connector 从来源队列读取一条消息，以执行一次破坏性读取（也就是说，它将从队列中删除消息）。

2.MQ Connector 将该消息写入工作队列。如果并行运行，则会存在多个工作队列，每个并行管道一个工作队列。MQ Connector 随后会提交本地事务。请注意，这里使用的本地事务表明，从来源队列到工作队列移动的消息是受保障的。如果移动失败，那么删除操作将会回滚，该消息会还原到来源队列上，该作业也会中止。

3.DTS 阶段将更新写入目标数据库。

4.DTS 阶段从工作队列中删除此消息。

通过使用工作队列，MQ 来源连接器能够并行运行。这个阶段的每个实例都从来源队列读取不同的消息，因为它以一种破坏性方式读取数据，而且 WebSphere MQ 可确保多个读取器中只有一个读取器能读取任何特定的来源消息。

在提交该事务时，如果一切正常，那么工作队列和数据库更新就会被提交，这些全部在单个 XA 事务中完成。如果出现任何故障，比如数据库写入错误，该 XA 事务将会回滚，这会撤销对目标数据库的任何更改，将 MQ 消息还原到工作队列。如果该作业随后重新启动，MQConnector 首先从工作队列读取消息，并将它们提供给该作业，然后继续从来源队列中读取数据。这意味着在发生故障时，重新启动作业即可继续从上次的位置开始处理消息。

本地事务与 DTS

除了处理符合 XA 的资源（比如 DB2 和 Oracle）之外，DTS 还会处理不符合 XA 的资源。这包括 ODBC 和 Teradata 目标。本节将介绍如何在 DataStage 作业中利用不支持 XA 架构的资源，以及如何仍然实现有保障的传送。还将介绍这种设计模式的缺陷和优势。

TS 处理非 XA 资源的方式是，除了 XA 事务之外，创建一个或多个 “本地” 事务。在此上下文中，一个资源由连接属性惟一地定义，比如数据库和用于访问该数据库的用户名。如果有多个访问同一个资源（也就是同一个数据库，使用相同的用户凭据）的链接，那么可以跨这些链接使用单个事务。在提交时，本地事务会在 XA 事务之前提交。这次本地提交可能获得成功，这是最可能发生的情形，也可能会失败，这会导致资源回滚该事务。在提交或回滚本地事务后，就会提交或回滚 XA 事务。在结合使用本地事务和 XA 事务时，可能的提交或回滚场景包括：

1.地提交成功，然后 XA 提交成功（正常的情况）。

2.本地提交成功，但 XA 提交失败（极少发生）。

3.本地提交失败。在发生这种情况时，DTS 会回滚 XA 事务。如果在回滚 XA 事务之前，DTS 发生崩溃或发生了其他某种系统故障，MQ 事务管理器会自动回滚它。

对于回滚：

1.本地回滚成功，然后发生 XA 回滚。

2.本地回滚失败。此场景不会实际发生；资源会假设一个回滚位置，除非系统要求提交它们，并且可以成功完成该提交请求。

在这些场景中，惟一要担忧的是第二种场景，也就是本地事务提交成功，但后续 XA 提交失败。在这种情况下，最糟的情形是数据在本地事务下被写入到目标数据库，但消息队列在来源队列上。如果出现这种情形，只要该作业是幂等的，就可以再次运行它。在前面描述的所有可能场景中，都不会丢失数据。下一节将介绍如何创建可在作业失败时重新启动的幂等作业。

幂等性和 DataStage 作业

本节将介绍如何编写作业，以便它们提供幂等的行为；也就是说，它们可在失败时重新启动，而且不会丢失数据。这些作业需要能够接受它们可能已处理的数据，要么再次处理它，要么忽略它。为了帮助解释此概念，可考虑一个非幂等的作业。考虑一个简单的文件到 Oracle 连接器作业，如图 3 所示：

图 3. 一个简单的文件到 Oracle 连接器作业

包含图 4 中指定的 Oracle 连接器的属性：

图 4. Oracle 连接器属性

假设表 TABLE_WITH_PK 拥有一个惟一约束，也就是说，它无法拥有两个具有相同键值的行。可以明显地看到，如果第二次运行此作业，它会失败，因为相同的数据将 “重放” 到 Oracle 连接器，这将导致重复键行错误。可通过多种可能方式克服此问题：

将 Write mode 属性更改为 Insert new rows only、Insert then update 或 Update then insert。第一个选项忽略数据库中已存在的行，第二个选项在该行上执行一次插入操作，如果返回结果表明已存在该行，则会使用相同数据执行一个更新语句。最后一个选项与此类似，但按照相反顺序执行该语句。

向目标阶段添加一个拒绝链接并配置它，以便导致行错误的记录被发送到该拒绝链接。

使用一个稀疏查找阶段查看一条记录是否已存在于目标中。查找的结果可用于确定是将记录发送到目标阶段、忽略它们，还是将现有记录转移到一个日志文件或其他目标。

具体使用哪种方法取决于特定的用例，以及是否需要对源数据进行处理。如果需要对源数据执行转换或使用其他功能，那么可以在作业中尽早使用一个稀疏查找阶段，以确定某条记录是否已存在，这样做会更有效，可以跳过转换。如果大部分源数据已通过少量中间阶段直接传递到目标，可能允许数据到达目标连接器，并让该连接器拒绝或忽略该数据会更有效。

通过使用额外的目标表来存储事务状态，可完成实现幂等性的其他方式。为此，会要求一个目标阶段可在单个事务中写入多个目标表。数据库连接器拥有这样一项功能，此功能将在下一节中介绍。

本地事务与数据库连接器

从 Information Server 8.5 开始，数据库连接器就已提供了支持多个输入链接的能力，每个链接以一个特定的数据库表为目标，拥有自己的一组属性（比如写入模式）。这些链接都在同一个数据库事务中执行，所以需要使用 ACID 功能来提供有保障的传送：要么写入所有表更新，要么全都不写入。

该功能有许多用例，包括：

写入相关的记录，比如在一个事务中保持父子关系。

对每个链接使用不同的写入模式，其中每个链接被配置为更新同一个表。例如，链接 1 可能从一个表执行删除，而链接 2 对同一个表执行更新。

存储一个事务标记来保存事务的状态。

后一种方法常常用在检查点重新启动 场景中，以避免在一个作业由于某个失败条件而需要重启系统时，将相同的数据再次发送到目标数据库。实现此目的的一种做法是，利用一个目标表来存储最后处理的行的编号。因为这是在与目标更新相同的事务中提交的，所以可保证准确地反映了最后处理的 行计数。在同一个作业的早期阶段，一个查找阶段会读取这个最后处理的 值，将它传递给一个转换器阶段，该阶段包含一个约束表达式，用于对比该行号与该计数。此阶段将会丢弃任何行计数小于最后处理的 值的行，因为这些行已由目标阶段处理。结果是，只有还未被目标阶段处理的记录被传送到该阶段。

还需要一个阶段，即 Wave Generator 阶段。在配置了多个输入链接时，数据库连接器仅在每个批行结束时提交该事务。Wave Generator 提供了多种不同方法来确定何时发出一个批次标记。对于本例，一个绝对行计数就足够用了，这样目标连接器就会每隔 N 行提交该事务一次。

完整的作业类似于图 5：

图 5. 检查点重新启动作业

像 图 5 中这样的作业可用作幂等性问题的解决方案，而且可确保将数据从来源传送到目标。

其他设计模式

本节将介绍可对其他资源实现有保障传送的其他一些模式。

使用 InfoSphere Data Replication 执行有保障的传送

将 Change Data Capture Transactional 阶段 (CDCTS) 与 InfoSphere Data Replication (DR) 结合使用，可实时复制和处理任务关键型数据事件。DR 使用一种基于日志的捕获方法来检测对各种数据库系统的更新，然后将这些更新复制到另一个数据库系统，或者将这些更新传送到文件、WebSphere MQ 消息或 CDCTS。CDCTS 通过 TCP/IP 与 DR 进行通信，以接收这些对来源数据库的更新，并传递状态消息。

图 6 显示了一个简单的 DR DataStage 作业：

图 6. 数据复制作业

该 CDC 阶段配置了 DR 订阅的名称，必须拥有至少两个输出链接：其中一个或多个链接承载了数据库更新，而只有一个链接负责传递书签。书签类似于上一节中介绍的最后处理的 计数。它用于识别当前事务的最后提交点。一个 CDCTS 作业使用一个目标数据库连接器，该连接器拥有至少两个输入链接。一个链接承载书签值，它的目标是一个简单表，也就是为了存储书签值而创建的书签表。与前一个例子一样，目标数据库连接器在事务批次结束时提交该事务。这些事务批次的边界由 DR 确定。DR 将提交消息发送到 CDCTS，这些消息与来源数据库的更新事务边界保持一致。在 CDCTS 收到这些提交消息后，它将事务批次标记发送到其输出链接。在目标数据库阶段收到每个链接上的批次标记时，它会提交该事务。因为一个事务同时包含书签表和数据库更新，所以存储的书签值可保证与数据库更新保持同步。

在作业运行时，DR 定期要求 CDCTS 报告最后提交的书签值。CDCTS 通过一个 ODBC 连接查询书签值，并向 DR 报告该书签。DR 使用该值清理日志，因为它可以确定一个特定的事务已成功写入到目标表。

在作业、网络或系统发生故障时，该作业会保留为不完整状态。在作业重新启动时，CDCTS 向 DR 报告最后提交的书签值，DR 随后使用该信息确定从哪条记录开始。

从 DR 到目标数据库的传送再一次受到了保护，而且通过使用书签机制，绝不会有重复的记录发送到目标数据库。

使用管道阶段执行有保障的传送

作业还可通过其他有创意的方式写入，以保证有保障的传送。大多数方式都依靠事务批次标记的使用，如前面的示例中所示。这类使用的另一个示例是，数据必须从一个来源队列移动到目标数据库，但消息系统不是 WebSphere MQ。DTS 仅支持 WebSphere MQ，所以要使用其他消息系统（比如 Java Messaging System (JMS)）实现类似功能，还需要使用一个类似图 7 中所示的作业：

图 7. 管道作业

SourceMessage 和 DeleteMessage 阶段使用了一个在 Java Integration Stage 基础上编码的 JMS 解决方案。这个 JMS 阶段也定期输出批次标记。该解决方案之所以有效，是因为 Teradata 连接器提供了将成功的记录发送到它的拒绝链接（如图 8 所示）的能力（只要在拒绝链接属性中被配置为这么做）：

图 8. Teradata 连接器拒绝属性

该作业执行过程中的事件顺序是：

1.JMS 阶段读取来源消息，但将它留在来源队列上。这个阶段还按规定的间隔发出批次结束标记。

2.来自 JMS 的数据由 Teradata 连接器写入到 Teradata 数据库中。如果写入成功，那么连接器会将该数据转发到它的拒绝链接，该链接被配置为仅转发成功的记录。

3.在 Teradata 连接器从 JMS 来源阶段收到批次结束标记时，它会提交事务，并将批次结束标记转发给它的输出链接。

4.当 Remove Duplicates 阶段收到批次结束标记时，它会删除重复数据并将结果（包含一个批次结束标记）发送给它的输出链接。

5.最后的 JMS 阶段在收到一个批次结束标记时删除来源消息。

请注意，如果没有批次结束标记，前面的解决方案将无法正确工作，因为 Teradata 在数据写入后而不是在提交后将记录发送到它的输出链接。但是，通过使用批次标记并包含 Remove Duplicates 阶段，该顺序得到了保障。

在数据库事务提交后，作业可能会失败。在这种情况下，不会执行来源消息的最终删除，该消息会保留在来源队列上。没有数据会丢失，但这种失败可能意味着如果改作业重新启动，Teradata 连接器会看到它已处理的相同数据。出于这个原因，与使用本地事务和 DTS 一样，作业设计需要是幂等的。使用正确构建的幂等性作业，如果作业因为任何原因而被中止，那么可以重新启动它，从离开的位置继续处理。