一种数据流程的处理方法和系统

摘要

一种数据流程的处理方法，基于逻辑上能够互联互通的节点形成的网络，该方法将一个任务对应的数据处理流程划分为多个并行数据处理子流程，以及确定所述数据处理流程涉及的节点，在所述数据处理流程的源节点和目的节点之间，为每一个数据处理子流程创建对应的数据管道，所述节点按照数据处理子流程的顺序，利用对应的数据管道完成每一个数据处理子流程的数据处理操作。该方法能够最大限度地提高数据处理子流程的并行程度，加速数据流程的处理，降低节点缓存数据所需的存储空间。本发明还提供一种数据流程的处理系统。

权利要求

1.一种数据流程的处理方法，基于逻辑上能够互联互通的节点形成的网络，其特征在于：将一个任务对应的数据处理流程划分为多个并行数据处理子流程，以及确定所述数据处理流程涉及的节点；

在所述数据处理流程的源节点和目的节点之间，为每一个数据处理子流程创建对应的数据管道；

所述节点按照数据处理子流程的顺序，利用对应的数据管道完成每一个数据处理子流程的数据处理操作；

所述数据是能够独立处理的最小数据元组；

其中，所述方法还包括：将每一个数据处理子流程待处理的数据通过对应的管道读入到本地存储器暂存，以及，在每一个数据处理子流程的数据处理结果输出前将所述结果暂存到本地存储器；

节点判断本地的数据处理能力是否超过指定阀值，如果超过，选择本节点的分流节点，为所述分流节点分配数据处理任务涉及的能够分流的数据处理子流程，在本节点和分流节点之间，以及在分流节点和目的节点之间，为能够分流的数据处理子流程创建对应的数据管道，所述分流节点按照数据处理子流程的顺序，利用对应的数据管道完成每一个数据处理子流程的数据处理操作。

2.如权利要求1所述的方法，其特征在于：用不同的线程分别完成待处理数据的输入、数据处理和数据处理结果的输出。

3.如权利要求2所述的方法，其特征在于还包括，为每一个数据处理子流程创建对应的数据处理线程，以及，所述数据处理线程并行处理对应的数据处理子流程待处理的数据。

4.如权利要求1所述的方法，其特征在于还包括：节点判断本地的数据是否有错，如果有，重新创建本节点与源节点和目的节点之间的数据管道，利用所述管道重新加载数据，从断点继续处理数据流程。

5.如权利要求4所述的方法，其特征在于：源节点在连接目的节点的数据管道处于空闲状态时，将数据处理子流程的操作结果写入该管道，通过该管道传输到目的节点。

6.如权利要求1所述的方法，其特征在于，所述管道是网络共享内存区域或网络通信信道。

7.如权利要求1所述的方法，其特征在于，利用SOCKET连接建立所述管道或通信信道。

8.一种数据流程的处理系统，包括多个逻辑上能够互联互通的网络节点，所述节点是一个任务对应的数据处理流程涉及的节点，所述数据处理流程包括多个并行数据处理子流程，所述节点包括用于执行所述数据处理子流程的数据处理单元，其特征在于：还包括：在所述数据处理流程的源节点和目的节点之间，为每一个数据处理子流程创建的数据管道，所述数据处理单元按照数据处理子流程的顺序，利用对应的数据管道完成每一个数据处理子流程的数据处理操作；

所述数据是独立处理的最小数据元组。

所述节点还包括：

用于将每一个数据处理子流程待处理的数据通过对应的数据管道读入到本地存储器的输入控制单元，以及，用于输出暂存到本地存储器的数据处理结果的输出控制单元；

负载均衡单元，判断本地的数据处理能力是否超过指定阀值，如果超过，选择本节点的分流节点，为所述分流节点分配数据处理任务涉及的能够分流的数据处理子流程，向分流节点和目的节点发出指示，以在本节点和分流节点之间，以及在分流节点和目的节点之间，为能够分流的数据处理子流程创建对应的数据管道，以及，指示所述分流节点按照数据处理子流程的顺序，利用对应的数据管道完成每一个数据处理子流程的数据处理操作。

9.如权利要求8所述的系统，其特征在于，所述节点还包括：

用于暂存每一个数据处理子流程待处理数据和处理结果数据的数据缓存单元，以及，用于管理所述待处理数据和处理结果数据的缓存控制单元。

10.如权利要求8所述的系统，其特征在于，所述节点还包括：

错误处理单元，判断本地的数据是否有错，如果有，重新创建本节点与源节点和目的节点之间的数据管道，以利用所述管道重新加载数据，从断点继续处理数据流程。

11.如权利要求8-10任一所述的系统，其特征在于：数据处理单元包括为每一个数据处理子流程创建对应的数据处理线程，所述数据处理线程并行处理对应的数据处理子流程需要处理的数据。

12.如权利要求8所述的系统，其特征在于，所述管道是网络共享内存区域或网络通信信道。

说明书

一种数据流程的处理方法和系统

技术领域

[0001] 本发明涉及一种数据流程的处理方法和系统。

背景技术

[0002] 在涉及多个数据处理节点的网络中，每个节点通常用于完成不同的数据处理操作，因此希望这些节点并行工作以提高数据处理的效率。例如，在一个用于数据仓库建立的ETL(抽取转换加载，Extract、Transform、Load)系统中，ETL数据流程处理逻辑架构包括负责数据流控制的数据总线(Data Bus)和多个负责数据处理的ETL节点，每个节点用于完成ETL操作中的不同数据处理功能，如果这些节点并行工作，将会极大提高ETL操作的效率。通常，一个独立的ETL流程对应于数据总线上的一个数据处理流程，多节点的并行数据处理指的是将顺序串行执行的数据流程转换为多个并行执行的数据处理流程。在保证数据并行处理的前提下，并行数据流程处理过程必须实现数据的完整性，流程的可靠传输以及数据流断点恢复的能力。

[0003] 为完成并行数据处理任务，目前的并行数据流程处理方案提供了一种节点内并行、节点间串行的数据处理方法，该方法在ETL数据总线之上创建多个执行逻辑上并行的数据处理子流程，在参与并行处理的数据处理节点，对应各个数据子流程，创建独立的线程处理数据，节点上每一个线程对应一个串行传输的数据流。在这种方式下，其中一个串行传输数据流中的数据以批次为单位传输，由于节点内的所有线程在并行工作，可以提高数据处理的速度，但是在节点之间，只有当一个节点上的所有线程都完成数据处理后，才会以数据包的形式将所有线程的数据处理结果发送到下一个节点处理。

[0004] 可见，采用上述并行数据流程处理方案分批处理数据会造成在A节点处理和传输数据的过程中，其后续节点B对应的处理线程处于等待状态，降低了流程的整体数据处理效率。而且，每个节点在处理数据之前需要存储当前处理各个线程需要处理的所有批次数据，占用大量的存储空间。另外，在数据流端点恢复处理功能实现上，由于数据是分批次处理的，断点恢复只能以节点的数据批次为单位。某批次数据处理错误发生故障只能以重新执行该批次数据数据处理流程实现断点恢复。也就是说，一个节点中的所有线程中，只要其中的一个线程出现数据处理的故障，就会导致其它线程数据处理结果的延迟处理，从而导致节点整体数据处理的延迟，这不可避免地降低整个流程的数据处理效率。

发明内容

[0005] 针对现有的并行数据流程处理方案的缺点，本发明要解决的问题在于，提供一种能够节省节点存储资源、具有较高数据处理效率的数据流程处理方法和系统，实现多节点网络的快速数据处理。

[0006] 本发明实施例提供的数据流程的处理方法，基于逻辑上能够互联互通的节点形成的网络，该方法首先将一个任务对应的数据处理流程划分为多个并行数据处理子流程，以及确定所述数据处理流程或子流程涉及的节点；基于所述节点和多个并行数据处理子流程，在所述数据处理流程的源节点和目的节点之间，为每一个数据处理子流程创建对应的数据管道，利用对应的数据管道，所述节点按照数据处理子流程的顺序，完成每一个数据处理子流程的数据处理操作。

[0007] 对于上述方法一种优选的方式是：分别用不同的线程分别完成待处理数据的输入、数据处理和数据处理结果的输出。

[0008] 本发明实施例提供的数据流程的处理系统，包括多个逻辑上能够互联互通的网络节点，所述节点可以是系统中一个任务对应的数据处理流程涉及的节点，所述数据处理流程包括多个并行数据处理子流程，所述节点包括用于执行所述数据处理子流程的数据处理单元。在一个数据处理流程涉及的节点中，由于不同的节点用于实现不同的数据操作功能，节点之间相互协调才可能完成一个数据处理任务，因此，节点之间存在相对的源和目的关系，在系统的实施例中，根据所述源和目的关系，在源节点和目的节点之间，还包括为每一个数据处理子流程创建的数据管道，所述数据处理单元按照数据处理子流程的顺序，利用对应的数据管道完成每一个数据处理子流程的数据处理操作。

[0009] 对于上述系统，一种优选的方式是：分别用不同的数据处理单元分别完成待处理数据的输入、数据处理和数据处理结果的输出。

[0010] 按照本发明实施例提供的数据流程的处理方法和系统，由于在一个任务对应的数据处理流程中涉及的有相对关系的源节点和目的节点之间，设置有为每一个数据处理子流程创建的数据管道，这样，所述数据处理单元就可以按照数据处理子流程的顺序，利用对应的数据管道独立地完成每一个数据处理子流程的数据处理操作，由于数据处理子流程之间是并行的关系，相互的数据处理不相关；借助各自的数据管道，一个节点内的数据处理子流程能够独自完成数据输入、处理和输出的任务，彼此不相互影响，因此能够最大限度地提高数据处理子流程的并行程度，加速数据流程的处理速度，降低节点缓存数据所需的存储空间。尤其是，待处理数据的输入、数据处理和数据处理结果的输出用不同的线程完成，使得数据传输和数据处理过程同时进行，特别是在网络数据传输延迟不可避免的情况下使得数据处理和传输在时间上最大限度重叠，更能达到提高数据流处理速度的目的。

[0011] 本发明的其它优点在后续的文字中有详尽的叙述。

附图说明

[0012] 图1是网络节点的逻辑关系图；

[0013] 图2是本发明所述方法的实施例的流程图；

[0014] 图3是图1所述实施例的原理图；

[0015] 图4是说明图1所示实施例的步骤21涉及的数据处理子流程划分原理图；

[0016] 图5是节点间负载均衡处理示意图；

[0017] 图6是本发明所述系统实施例逻辑图；

[0018] 图7是图6所述系统实施例采用节点实施例示意图。

具体实施方式

[0019] 在实际的涉及多个数据处理节点的网络中，无论节点所处的物理位置如何，也无论节点之间物理上的连接关系如何，业务相关的节点之间形成的网络结构逻辑上都可以看作能够实现节点互联的总线型结构。例如，图1中，节点A到节点N是业务相关或任务相关的节点，节点之间的顺序或工作关系由一个具体的任务流程确定，它们物理上可以是跨域分布的，但逻辑上都可以连接到一个数据总线并通过该总线互联起来，这个数据总线是逻辑总线，其中可以包含多条逻辑数据通道，等等。图1中的节点逻辑上都能够通过各自的适配器等与数据总线连接，因此可以实现互联互通。图1中，节点A到节点N在一个具体的任务流中完成的功能不同，是不同的功能节点，这些节点完成具体任务执行的数据处理流程可能需要调用网络中的其它资源，如图1中计算网络的资源、数据源A到数据源N的资源，以及集结库资源等等，数据处理结果也可以存储到节点以外的其它节点，例如数据仓库等。

[0020] 图1所示网络只是一个简图，其中还隐含其它节点，例如涉及系统服务的用于任务制定和管理的节点，对一个任务进行过程监控的节点等等。本发明实施例实现的数据处理的并行，通过减少数据处理子流程彼此之间的影响，并利用时延隐藏原理实现数据处理过程在时间上的最大限度的重叠来实现快速的数据流处理。因此，本发明尤其适合应用于ETL系统的数据处理。

[0021] 下面参照附图对本发明的实施例作详细说明。

[0022] 图2是本发明所述方法的第一个实施例流程图。图2所述实施例的原理参考图3。图3可以看作图1所示网络的衍生实例，一个ETL系统的逻辑模型，图2所示实施例以图3所示ETL系统为例说明。图1中的部分节点被抽象为图3中系统层方框33中的系统服务节点和ETL模型服务节点；部分节点被抽象为图3中组件层方框34中的任务引擎服务A节点、任务引擎服务B节点、转换服务节点、清洗服务节点和增量捕获服务节点；部分节点被抽象为图3中资源层方框35中计算服务节点和源数据服务节点。图3描述了从三个业务库A、B、C向数据仓库36导入数据的过程。

[0023] 按照图2所述实施例，在步骤21将一个任务对应的数据处理流程划分为多个并行数据处理子流程，以及确定所述数据处理流程或子流程涉及的节点。其中，所述任务由一个图3中或图1中没有绘出的用于任务制定和管理的节点完成，例如ETL系统中的一个的用于构建ETL任务模型的节点。该节点创建了一个具体的解决某个数据处理问题的任务，该任务对应的数据处理流程包含多个数据处理子流程(图3中的方框32)，即，将该任务对应的数据处理流程划分为多个并行的数据处理子流程，其中的某个数据处理子流程还可能包含另一个数据处理子流程31，即数据处理子流程可能具有嵌套关系，这根据数据处理的复杂度决定。创建任务及其对应的数据处理流程，以及将所述数据处理流程划分为多个并行的数据处理子流程的过程属于公知技术，例如ETL领域的公知技术，此不赘述。当所述数据处理流程或子流程确定后，根据该任务需要完成的操作，就可以通过计算获知所述数据处理流程或子流程涉及的节点，由于确定节点的过程也属于公知技术，此也不赘述。

[0024] 按照图3的示例，数据处理流程32包含的多个并行数据处理子流程涉及的组件层节点包括2个任务引擎服务节点和转换服务、清洗服务、增量捕获服务节点各一个，各个组件层组件以协同的方式处理数据流程，从而完成数据流的执行。在任务执行过程中，上述节点根据数据处理子流程的需要可以调用资源层的节点，获得所需的数据服务，也可以与系统层节点进行互访调用，为其它节点提供数据服务。

[0025] 步骤21所述数据处理子流程的划分根据任务的不同可能有多种划分形式，这取决于系统的数据属性和任务的要求等等，尽管这属于公知技术的范畴，为了更好地说明本实施例，下面举例说明步骤21。参考图4，数据处理流程的划分指的是在保持数据完整性的前提下将单个ETL任务中的数据流分割为多个并行处理的独立数据流程，以实现多个数据流的并行处理。图4(a)描述了把从表DS_RELATIONS中REAL_FLOAT和IMAGE字段中抽取出来的数据处理流程划分成了两个并行的子流程。图4(a)描述划分数据处理流程的方式，其前提是假设数据处理的对象是数据元组(Tuple)，这种划分不改变抽取数据元组的结构，把单个数据处理流程的操作划分到两个子流程处理。图4(b)描述了根据转换定义，分别把从表DS_RELATIONS中NAME，AGE字段中抽取出数据和从ARTICLE，MODIFY_TIME字段中抽取出数据的数据处理流程划分为两个独立的数据处理子流程并行执行。图4(b)描述的划分方式是按照ETL流程中聚合、组合、清洗等转换规则的定义和与目的端数据库表中字段的映射定义，将单个数据元组划分为多个独立的数据元组的数据处理流程划分为两个并行处理的子流程。通常情况下需要结合两种划分方式实现ETL任务数据流的分割。需要说明的是，对于不同任务的数据处理流程，数据处理流程的划分规则在任务模型中定义，并交由具体的执行数据处理操作的节点执行。

[0026] 还需要说明，所述数据处理子流程的划分可以在任务制定和管理的节点完成，也可以在所述数据处理流程或子流程涉及的节点完成，这取决于具体的系统设计，由于这属于公知的范畴，此不赘述。另外在数据处理流程或子流程涉及的节点开始执行时，需要向节点注册节点任务模板，通常，每一个数据处理流程对应一个任务模板，所述任务模板可以采用文件的形式描述，例如XML文件。所述任务模板包含的信息主要考虑系统赋予节点的任务或节点的功能考虑，这在不同的系统中有很大差异，例如：

[0027] 1)系统中可用节点定义：如节点物理地址、节点服务接口的定义描述。这些信息用于负载均衡过程中查找计算资源分流节点和建立节点之间数据管道用。

[0028] 2)本节点数据处理任务定义：如数据转换、清洗规则定义。用于描述不同数据处理流程中数据处理方式。

[0029] 3)数据流定义：如数据处理流程或子流程涉及的各个数据管道(ID、类型、名称、描述等属性)定义，先序、后序节点定义，管道安全属性定义，等等。

[0030] 图2的步骤22，在所述数据处理流程的源节点和目的节点之间，为每一个数据处理子流程创建对应的数据管道。所述数据管道根据系统设计的不同，第一种情况，所述数据管道可以是网络共享内存区域，物理上由创建于管道两端节点之上的分布式一致性的高速缓存和Socket连接建立的网络通信信道构成，因此也可以理解为两端节点都可以访问的同步虚拟网络存储区。第二种情况，所述数据管道也可以是网络通信信道或获得系统虚拟总线控制权的时间片，等等。对于第一种情况，数据管道可以是虚拟网络存储区，也可以是有固定网络地址的用于存储数据的服务器地址，这种情形下的数据管道可以用于传输数据、存储数据，等等；对于第二种情况，数据管道就是一个单纯传输数据的通道，即为数据处理子流程提供用于传输数据的即时数据通道。

[0031] 对应地，数据管道的建立也有多种公知的方式，例如，在网络中的某个承担数据存储任务的节点建立一个特定的存储区，利用SOCKET连接建立所述管道，等等。如果采用第一种形式的数据管道，所述数据管道可以看成是用来完成节点之间数据输入管理、数据输出管理进程通讯的共享虚拟内存区域，可以将待处理数据和处理结果数据存储在所述数据管道，因此，在数据处理过程中，考虑到数据的性质、数据处理的独立性、数据处理速度、错误恢复的难易等因素，所述管道中的数据最好定义为可被独立处理的最小数据元组。

[0032] 具体的数据管道创建过程也有多种方式，数据处理流程中的所有相对的源节点和目的节点都可以发起创建管道操作。例如，在数据处理流程初始构建时源节点发起管道创建操作，如果节点采用每一个数据处理子流程对应一个包括数据处理线程、输入控制线程和输出控制线程的线程组的数据处理方式，通过创建本地输出控制线程并主动连接后序节点输入控制线程的方式创建数据管道。首先，数据管道创建发起节点向目的节点发送包含数据处理流程标识(ID)和管道ID的管道创建请求消息，基于该消息，目的节点查找本地是否存在数据处理子流程对应的线程组和数据管道，如果不存在，则基于本地存储器中对应ID的数据处理流程定义创建数据管道和对应的线程组，然后向数据管道创建发起节点返回请求已经执行完成的消息；如果存在，直接向数据管道创建发起节点返回请求已经执行完成的消息。本例中，采用SOCKET技术创建管道，创建管道需要的参数主要包含数据管道Socket连接所需要的先序、后序节点关联和数据管道安全属性等。

[0033] 对应地，如果节点采用多个数据处理子流程对应一个包括数据处理线程、输入控制线程和输出控制线程的线程组的数据处理方式，目的节点查找本地是否存在数据处理子流程对应的数据处理结构、数据处理方式等参数，以及创建相应的参数和数据管道即可。

[0034] 在图2的步骤23，所述节点按照数据处理子流程的顺序，利用对应的数据管道完成每一个数据处理子流程的数据处理操作。该步骤的简单数据处理方式可以是：每一个数据处理子流程的输入数据、输出数据在对应的数据管道完成，各个子流程的数据处理排队可以参考公知的高效排队方式，此不赘述。这种简单的数据处理方式，不但能够保证每一个数据处理子流程的高效执行，当一个子流程的数据处理由于各种原因中断，会立即启动下一个子流程的数据处理，因此，能够保证数据处理的高效。但是，还有效率更高的数据处理方式，在基于本第一实施例的第二个实施例中，分别用不同的线程分别完成待处理数据的输入、数据处理和数据处理结果的输出，即，每一个数据处理子流程至少对应一个数据处理线程、输入控制线程和输出控制线程，根据数据的不同、处理流程不同等，一个数据处理子流程还可以对应多个数据处理线程、输入控制线程和/或输出控制线程。这样，使得数据传输和数据处理过程同时进行，在数据传输延迟不可避免的情况下使得数据处理和数据传输在时间上最大限度重叠，达到提高数据流处理速度的目的。

[0035] 为了进一步优化上述实施例，在基于所述第二实施例的第三实施例中，为提高数据处理阶段的效率，为每一个数据处理子流程创建对应的数据处理线程，以及，这些不同的数据处理线程并行处理对应的数据处理子流程待处理的数据。

[0036] 还可以通过本地存储器的使用进一步优化本发明。在基于所述第三实施例的第四实施例中，将每一个数据处理子流程待处理的数据通过对应的管道读入到本地存储器暂存，以及，在每一个数据处理子流程的数据处理结果输出前将所述结果暂存到本地存储器。在所述第四实施例中，如果采用第一种形式的数据管道，不但数据处理子流程的待处理数据可以预先写入对应的数据管道，源节点在连接目的节点的数据管道处于空闲状态时，将数据处理子流程的数据处理结果写入该数据管道，通过该数据管道传输到目的节点。具体地，通过判断数据管道是否可写入，或者是否有存储空间等，获得所述数据管道是否处于空闲状态。

[0037] 在一个具体的数据处理流程中，数据处理流程中参与数据处理的节点输入、输出处理线程查找数据处理流程中涉及的节点的物理地址，相互关联的先序、后序节点之间分别创建数据管道，以及启动各个数据管道两端对应的数据处理线程。相对的源节点输出控制线程在向目的节点发送数据之前查找待发送的数据元组，并判断后序节点对应管道是否处于可写入的状态，如果处于不可写入状态，等待预定时间后继续判断，否则发送数据元组，在本地清除已经发送的数据元组。目的节点的输入控制线程从对应的管道中接收数据元组并存储到本地的存储器中，等待本地的数据处理线程处理。

[0038] 在基于上述实施例的本发明所述方法的其它实施例中，还包括下述优化技术方案：

[0039] 1、如果某个节点的数据处理能力不足，例如数据处理速度过低、存储的数据量过少等，说明该节点的负载过大，此时可以调用其它节点分担负载。具体的过程是，节点判断本地的数据处理能力是否低于指定阀值，如果低于，选择本节点的分流节点，将所述分流节点作为本节点和目的节点间的中间节点，为所述分流节点分配数据处理任务涉及的能够分流的数据处理子流程，在本节点和分流节点之间，以及在分流节点和目的节点之间，为能够分流的数据处理子流程创建对应的数据管道，所述分流节点按照相应的数据处理子流程的顺序，利用对应的数据管道完成每一个数据处理子流程的数据处理操作。

[0040] 例如，如果数据处理能力用本地存储器能够存储的数据量来衡量，在各个节点之间数据传输和数据处理速度不匹配的情况下利用本地存储器缓存数据就可能出现存储能力不足的问题，如节点收到的前序节点发送过来需要处理的数据的速度大于本节点的处理速度的情况下，本节点缓存的待处理数据量会不断增加，最终超过存储极限。在缓存数据量超过指定阀值时通过调用系统中的节点服务，查询本节点的存储资源分流节点，并创建两节点之间的通信管道和处理线程实现节点之间的负载均衡。具体的负载均衡原理参考图5。图5说明的是，在节点A的数据处理能力不足时，即当前节点A缓存的到后续节点C的处于待处理状态的数据量超过指定的阈值，通过查找网络中其它节点存储资源冗余情况，找出存储资源冗余的节点B，创建节点A到节点B和节点B到节点C的数据通道，启动节点A、B、C之间的数据管道对应的数据输入、输出和数据处理线程，继续完成数据段处理，使得通过节点B分担节点A的部分数据处理子流程的数据处理任务。对应地，如果数据处理能力用节点的计算能力来衡量，则在本节点的计算能力不足时，则通过查找网络中其它节点计算资源的冗余情况，通过冗余的计算资源节点解决负载均衡问题。

[0041] 2、数据处理过程中如果出现节点错误需要恢复的情况，节点部署的数据输入、输出和数据模块重新从本节点的存储器中加载当前各自处理的数据，重新创建节点之间的数据管道，从断点继续处理数据流程，实现数据的断点恢复。一个节点是否存在数据处理错误，可以由节点本身发现，例如通过数据校验的方法；也可以由与其关联的其它节点发现，例如通过数据的相关性发现；也可以由网络中负责数据处理流程管理的节点发现，等等。假设错误由节点本身发现，则节点需要预先布置错误检查流程，通过该流程，节点判断本地的数据是否有错，如果有，重新创建本节点与源节点和目的节点之间的数据管道，利用所述管道重新加载数据，从断点继续处理数据流程。

[0042] 通常，所述错误种类大致可划分为以下三种：

[0043] 1)节点物理故障。故障节点在重新启动时，执行错误恢复过程。具体为：输入、输出控制和数据处理线程从本地的存储器中加载任务模板及其对应数据，重新创建数据流对应线程和数据管道连接，恢复缓存中的数据处理和发送过程。

[0044] 2)数据管道连接中断。节点输入、输出控制线程在发送、接收数据时会主动探查数据管道连接是否中断，如果中断，则不断探测数据管道的连接状态，如果在指定时间内连接状态未恢复，则挂起对应的输入或输出控制线程，被动等待其先序、后序节点发送管道创建请求。

[0045] 3)数据处理错误。当节点数据处理发生数据元组中数据格式、类型定义与数据处理规则不匹配或数据处理过程中产生异常，直接将数据元组对象状态标记为处理错误，并将异常编码写入元组对象保存到本地存储器中。此种错误需要人为干预处理错误元组数据，因此只需记录。

[0046] 在错误恢复过程中，发生错误恢复的节点会基于本地数据流定义主动查找先序、后序节点，发起管道创建过程，重新从本地存储器中查询数据，恢复错误发生的状态。

[0047] 下面通过一个具体实施例进一步说明本发明的方法。

[0048] 假设需要处理的数据元组格式定义包含Column_A和Column_B两个字段，所述数据需要首先经过清洗节点Node_Cleansing进行数据清洗，之后经过数据格式转换节点Node_Transform处理生成最终结果数据元组。

[0049] 对于每一个数据处理的子流程，源端的数据抽取节点的数据输出控制线程，按照任务定义查找数据流中参与数据处理的数据清洗节点(Node_Cleansing)和数据转换节点(Node_Transform)，并从任务定义中解析出两节点物理地址信息。依次创建源端数据抽取节点输出到Node_Cleansing输入的数据管道(PipeLine_1)，Node_Cleansing的输出到Node_Transform输入的数据管道(PipeLine_2)，Node_Transform的输出到其对应的目的端数据加载节点的输入的数据管道(PipeLine_3)。在各个节点的输入、输出控制线程和数据处理线程分别创建对应数据管道的线程监视数据管道和本地节点存储器中的数据元组。

[0050] 数据流源端的数据抽取节点从数据库中抽取出数据，并按照每个元组包含字段Column_A、Column_B最小单位数据生成元组数据包对象，在设置了流程标记ID后存储在抽取节点的存储器中，处理完毕后等待对应数据管道的输出控制线程读取并将元组对象直接写入到对应的数据管道中。对于输出控制线程，将数据处理线程处理完毕的结果数据写入数据管道的过程与写入本地存储器的过程相同。当数据抽取节点的输出控制线程判断数据管道状态为可写入，则将元组对象写入管道PipeLine_1后，数据会直接通过Socket连接发送到连接Node_Cleansing的数据管道，此时Node_Cleansing输入控制线程能够从管道中读取数据，将读取的元组对象状态设置为待处理，并写入本地存储器缓存。

[0051] Node_Cleansing的数据处理线程对应于该管道输入控制线程，在启动后不断查询本地存储器中指定流程标记ID且状态为待处理的数据元组，待数据处理完毕后再将数据处理结果记录在元组对象中并将对象状态标记为待发送。Node_Cleansing的输出控制线程则在启动后不断查询本地存储器缓存的指定流程标记ID且状态为待发送的数据元组，在管道PipeLine_2状态为可写入时将元组数据写入PipeLine_2并清除写入成功元组在本地存储器中的记录。Node_Transform节点处理过程与Node_Cleansing相同。

[0052] 在Node_Cleansing节点，例如缓存控制的线程发现本地存储器中某流程标记ID待处理状态数据量超过预定的阀值100MB时，则触发输出控制线程查找当前网络中其他节点资源的使用状态，找到资源冗余的节点Node_Cleansing_2并发送分配给节点Node_Cleansing_2的任务及相应的数据处理子流程ID。接着初始化节点Node_Cleansing_2，初始化过程包括任务的解析、创建相应数据处理线程等。之后创建对应流程标记ID的数据流程的输出控制线程和从Node_Cleansing节点到节点Node_Cleansing_2的数据管道PipeLine_2_1，节点Node_Cleansing_2到Node_Transform节点的数据管道PipeLine_2_2。启动数据管道PipeLine_2_1和PipeLine_2_2对应的各个线程完成负载均衡。

[0053] 本发明提供的数据流程的处理系统，参考图6。在图6所示的系统中，网络节点逻辑上能够通过一个物理或逻辑的数据总线互联互通，图6中的节点在一个具体的任务流中完成的功能不同，是不同的功能节点，实际中的节点可能是一台服务器，也可能是一个局域网，而且，这些节点完成具体任务执行的数据处理流程也可能涉及网络中的其它节点。

[0054] 为方便说明本实施例，可以将图6中的节点可视为一个任务对应的数据处理流程涉及的节点，且相互之间为源和目的的关系，所述数据处理流程包括多个并行数据处理子流程，所述节点包括用于执行所述数据处理子流程的数据处理单元。在图6中的源节点和目的节点之间，还有为每一个数据处理子流程创建的数据管道，通常，数据管道的数量与节点处理的数据处理子流程的数量相同。这样，节点的数据处理单元按照数据处理子流程的顺序，利用对应的数据管道完成每一个数据处理子流程的数据处理操作。很多时候，数据管道的数量与节点处理的数据处理子流程的数量不相同或不对应，对于部分数据处理子流程，也可以通过数据管道的复用构建本发明所述系统，鉴于复用技术是公知技术，此不赘述。

[0055] 图6中，所述数据处理子流程的划分可以在任务制定和管理的节点(图6未绘出)完成，也可以在所述数据处理流程或子流程涉及的节点完成，这取决于具体的系统设计，由于这属于公知的范畴，此不赘述。

[0056] 在本发明所述系统的其它实施例中，其中的节点还包括：用于暂存每一个数据处理子流程待处理数据和处理结果数据的数据缓存单元，以及用于管理所述待处理数据和处理结果数据的缓存控制单元。所述节点还可以包括：用于将每一个数据处理子流程待处理的数据通过对应的数据管道读入到本地数据缓存单元的输入控制单元，以及用于输出暂存到本地数据缓存单元数据处理结果的输出控制单元。具有这些技术特征的节点参考图7。

[0057] 上述系统的实施例中，所述数据管道具体的是网络共享内存区域，当然也可以是其它方式，因此所述数据管道可以用于传输数据、存储数据，等等。另外，在数据处理过程中，考虑到数据的性质、数据处理的独立性、数据处理速度、错误恢复的难易等因素，所述管道中的数据最好定义为可被独立处理的最小数据元组。

[0058] 在基于上述系统实施例的另一个实施例中，数据处理单元包括为每一个数据处理子流程创建对应的数据处理线程，所述数据处理线程并行处理对应的数据处理子流程需要处理的数据，这样更能提高系统的数据处理效率。

[0059] 在所述系统的另一个实施例中，所述节点还包括负载均衡单元，以图7为例，所述负载均衡单元与输入控制单元、输出控制单元、数据处理单元、缓存控制单元和数据缓存单元分别相连，用于判断本地的数据处理能力。具体地，所述负载均衡单元判断本地的数据处理能力是否超过指定阀值，如果超过，选择本节点的分流节点，为所述分流节点分配数据处理任务涉及的分支数据处理子流程，向分流节点和目的节点发出指示，以在本节点和分流节点之间，以及在分流节点和目的节点之间，为分支数据处理子流程创建对应的数据管道，以及，指示所述分流节点按照数据处理子流程的顺序，利用对应的数据管道完成每一个数据处理子流程的数据处理操作。有关负载均衡的其它说明可以参考本发明所述方法部分的实施例的说明。

[0060] 在所述系统的实施例中，还可以选择性地设置一个包括错误处理单元。所述错误处理单元如果设置在图7所示的节点中，则与输入控制单元、输出控制单元、数据处理单元、缓存控制单元和数据缓存单元分别相连，用于判断本地的数据处理是否存在错误，如果有，向相关单元发出指示，例如系统中的管理单元或其它执行相关操作的单元，重新创建本节点与源节点和目的节点之间的数据管道，以利用所述管道重新加载数据，从断点继续处理数据流程。有关错误处理的其它说明可以参考本发明所述方法部分的实施例的说明。

[0061] 实际上，本项发明的技术性范围并不局限于说明书上的内容，还有很多根据其权利要求确定的具体的技术性应用方案。以上所述仅是本发明的优选实施方式，应当指出，对于本技术领域的普通技术人员来说，在不脱离本发明原理的前提下，还可以作出若干改进和润饰，这些改进和润饰也应视为本发明的保护范围。