一种基于工作流的通用ETL工具的过程模型生成方法

技术领域

本发明属于数据处理技术领域，具体涉及一种基于工作流的通用ETL工具的过程模型生成方法。

背景技术

近几年，随着电子商务的崛起和信息技术产业的飞速发展，信息孤岛的存在变得越来越普遍。全球每年投入在应用系统的集成和数据整合有3000亿美元之巨，并且这个数据还在以每年28％的量在增长，同时，越来越火热的系统集成和数据仓库市场，使得越来越多的ETL工具涌现出来。

传统的ETL设计方法和ETL工具已经越来越不适应不断变化的需求，因为传统的ETL的设计和开发除了需要熟悉数据表的结构和处理数据的规则，以及非常好的编程能力之外，还需要对整个ETL的流程以及ETL流程的每一步的细节是如何处理的非常熟悉。当商业需求的规则或者数据源发生改变时，传统的ETL设计者必须重新编程设计整个ETL或者进行大量的修改工作。因此，对通用ETL的研究变得越来越迫切。它能够降低对ETL设计人员的要求，适应不断更改的业务逻辑，减少企业和单位的开发时间和成本，将数据仓库变得平民化。

目前，国内对通用ETL的研究几乎一片空白，一部分ETL工具是传统的对物化视图的维护，对设计者和使用者的要求非常高，一部分ETL工具是设计于专门的应用背景，只能在其特定的业务背景下使用，还有一部分则是直接编写脚本或程序来对源数据进行抽取、转换和加载，当数据的结构或ETL的需求发生改变时，整个项目几乎没有重用性可言。国外对ETL系统的研究相对而言要更加全面和成熟，ETL的研究主要体现在对整个ETL过程的建模以及ETL的转换过程和优化上。但是，无论上面提到的哪一种，都有几个共同的缺点，那就是可复用性差、通用性差。

发明内容

针对上述问题，本发明在传统ETL概念模型基础上，提出一种基于工作流的通用ETL工具的过程模型生成方法，对ETL概念模型进行抽象，完成整个ETL的控制以及任务的定时执行和调度，建立基于工作流的ETL模型工具，在此基础上构建ETL流程的有向图，同时集成变换、分解、合并、串行化、并行化、添加还原点、分流和复制八种State变换方式优化了状态转换。

实现上述技术目的，达到上述技术效果，本发明通过以下技术方案实现：

一种基于工作流的通用ETL工具的过程模型生成方法，其特征在于，包括以下步骤：

S1：建立基于工作流的ETL模型工具，包括作业主控模块和抽取、转换、装载模块；

S2：构建与所述ETL模型工具对应的流程图T，所述流程图T包括若干个有向图G，分别定义为G1…Gn；

S3：建立变换、分解、合并、串行化、并行化、添加还原点、分流和复制八种变换规则；

S4；在ETL模型工具整个流程执行过程中，利用所述八种变换规则来改变各有向图G，直至选出变换后时间最小的情况下的所有有向图G，构成通用ETL模型工具的过程模型，以实现优化ETL工具整个流程执行过程。

进一步地，所述步骤S1中的作业主控模块用于完成对ETL所有作业的调度工作，实现ETL作业的定时、排队、并行调度、流程控制、日志记录，根据配置文件确定各个Job的启动条件、执行顺序以及是否依赖于其他的Job，当某个Job满足启动条件时，作业主控模块启动该Job，并按照顺序执行该Job的子任务流程。

进一步地，所述子任务流程为transformation标签里的转换任务。

进一步地，所述步骤S1中的抽取、转换、装载模块中各个步骤之间的数据传输流程是一个生产者消费者模型，每一个节点对下一个节点而言是生产者，节点和节点之间采用一个阻塞队列传递数据，前一个节点根据其连线上的规则，将数据处理后写入阻塞队列，后一个节点一条一条的从阻塞队列里读取数据，所有的这些节点组成Transformation。

进一步地，所述步骤S2具体为：根据有向图的定义构建ETL模型工具的流程的有向图T＝(V，E)，其中V＝Step∪D；所述Step为Transformation的所有功能单元；所述D为Step之间的数据、关系和参数；所述E为有向图的边，实际就表示过程。

进一步地，所述步骤S3中的八种变换规则具体包括：变换、分解、合并、串行化、并行化、添加还原点、分流和复制；

所述交换Swap(v₁,v₂)具体为：将节点v₁和节点v₂相互交换；

所述分解Split(v₁₊₂,v₁,v₂)具体为：将节点v₁₊₂分解为v₁、v₂；

所述合并Merge(v₁,v₂,v₁₊₂)具体为：将节点v₁、v₂合并为v₁₊₂；

所述串行化Factorize(v₁,v₂,v₃,v₁₊₂)具体为：将处于不同的数据流中，但是类型相同的节点v₁、v₂合并为一个新的节点v₁₊₂；

所述并行化Distribute(v₁,v₂,v₃,v₁₊₂)具体为：将节点v₁₊₂分解为两个类型相同的节点v₁、v₂，让节点v₁、v₂处于不同的数据流中；

所述添加还原点AddRecoveryPoint(v,v_RP)具体为：在节点v处添加还原点v_RP；

所述分流Partion(v₁,v₂,n)具体为：将节点v₁、v₂之间的数据流分为n份；

所述复制Replicate(v₁,v₂,n)具体为：将节点v₁、v₂之间的数据复制n份。

进一步地，所述步骤S4具体为：使用穷举法将所述的八种变换规则都代入与所述ETL模型工具对应的流程图T，然后比较，选出变换后时间最小的有向图。

进一步地，所述比较、选出变换后时间最小的有向图的具体过程为：使用open和close两个状态队列，来存储未访问和已访问的状态，利用S_MIN来表示时间最小的状态，初始化时，让S_MIN等于初始状态S_G，open包含S_G，close为空，对open队列里的所有状态S，首先将状态S出队列，然后S与S_MIN进行比较，如果其时间小于S_MIN，那么S为新的最小时间状态，再让S根据所述的八种变换规则产生新的状态S'，如果S'既不属于open也不属于close，则将其送入队列open，然后将S送入close队列里，标记为已访问；不断重复上述步骤，直到open队列为空，从而实现优化整个ETL过程。

进一步地，在进行有向图的变换时，遵循以下规则：根据组成Transformation节点类型的不同，将Transformation总体上分为四类：普通Step组成的Transformation、包含分流节点的Transformation、包含还原点的Transformation、包含复制节点的Transformation；

(1)对于普通Transformation，其包括Swap、Split、Merge、Factorize和Distribute，只做相邻有向图之间的变换；

(2)对于包含分流节点的Transformation，在有并行化需求时，将DataStream分开在不同的物理处理器上处理，也就是所谓的分流或者并行处理；

(3)对于包含还原点的Transformation，当发生错误时，将当前的阻塞队列里的数据全部拷贝到硬盘里，将距离错误位置最近的还原点的数据读入程序，并且从这个还原点开始执行；由于并没有执行还原点之前的流程，所以从还原点开始执行要比重新执行整个流程要快；

(4)对于包含复制节点的Transformation，当对ETL流程的容错性要求高，而对效率的要求宽松时，采用复制ETL数据流，并将其多次执行的方法来用执行效率换取容错性。

本发明的有益效果：

本发明的基于工作流的通用ETL工具的过程模型，不同于传统的ETL模型，它屏蔽了底层的具体数据抽取、转换和加载步骤，当业务发生改变时，用户只需要修改基于工作流的ETL模型工具即可，通过多种ETL State变换方式使得整个ETL过程在经过恰当的变换之后，变成一个执行效率更高的新ETL过程，大大提高了开发效率，减少开发成本，真正的实现了数据仓库和数据集成平民化。

附图说明

图1为本发明一种实施例的结构示意图；

图2为ETL过程优化示意图；

图3为三种不同的流程处理示意图；

图4为交换规则的处理流程图；

图5为分解和合并规则的处理流程图；

图6为串行化和并行化规则的处理流程图；

图7为添加还原点规则的处理流程图；

图8为分流规则的处理流程图；

图9为复制规则的处理流程图。

具体实施方式

为了使本发明的目的、技术方案及优点更加清楚明白，以下结合实施例，对本发明进行进一步详细说明。应当理解，此处所描述的具体实施例仅仅用以解释本发明，并不用于限定本发明。

下面结合附图对本发明的应用原理作详细的描述。

一种基于工作流的通用ETL工具的过程模型生成方法，包括以下步骤：

S1：在传统ETL概念模型基础上，建立基于工作流的ETL模型工具，包括作业主控模块和抽取、转换、装载模块；

所述作业主控模块用于完成对ETL所有作业的调度工作，实现ETL作业的定时、排队、并行调度、流程控制、日志记录，根据配置文件确定各个Job的启动条件、执行顺序以及是否依赖于其他的Job，当某个Job满足启动条件时，作业主控模块启动该Job，并按照顺序执行该Job的子任务流程，即<transformation>标签里的转换任务；

所述抽取、转换、装载模块中各个步骤(即：抽取、转换、装载步骤)之间的数据传输流程是一个生产者消费者模型，每一个节点对下一个节点而言是生产者，节点和节点之间采用一个阻塞队列传递数据，前一个节点根据其连线上的规则，将数据处理后写入阻塞队列，后一个节点一条一条的从阻塞队列里读取数据，所有的这些节点组成Transformation。

具体参见图1，其中：

JOB(J)：负责对整个ETL的控制以及任务的定时执行和调度，其包含一个或多个Transformation；所谓作业流指的是当一个Transformation成功执行后，能够自动的执行其他作业；比如，当希望Transformation t1执行完毕之后，ETL的JOB会自动执行Transformation t2，则t1就被称为是上游作业(Up Stream Transformation)，t2则被称作下游作业(Down Stream Transformation)，是被自动执行的作业。

JOB的操作包括：

◆TableExists：用于判断用户指定的表是否存在，可以存在若干个Table。

◆FileExists：用于判断用户指定的文件是否存在。

◆Mail：用于对用户进行事件通知。

所述的Transformation(T)：是指数据抽取、转换、加载所形成的流程，完成针对数据的抽取、基础转换和数据加载，一个Transformation由多个Step组成。

所述的Step(S)：是Transformation的功能单元，用来完成特定的转换步骤，Step具体包括以下操作：

数据抽取：

TableInPut：当MySQL、Oracle、SqlServer等数据库里的数据作为数据源的时候，系统与数据库采用JDBC/JNDI来连接。

FileReader：当源数据为文本文件的时候，需要专门的FileReader来对文本文件进行处理；当文件为Excel或XML文件时，采用Jdom来处理。

FTP：从FTP服务器中读取数据；

HTTP/WEB SERVICES：发送Web Service请求，然后从Web Service返回的数据流中读取数据；

SMTP(简单的邮件传送协议)：访问电子邮件服务器，根据用户名和密码从电子邮箱中读取数据；

企业级别JAVABEANS(EJB)：调用EJB接口，从返回的数据流中读取数据。并通过EJB接口，将数据传递到转换单元。

转换：

Joiner：实现两个数据集的Join操作，包括数据库中常见的连接类型，及数据库中未提供的部分连接类型，利用Joiner组件，还可以实现不同数据源的Join操作；

Filter：根据过滤规则对数据进行过滤；

Switcher：根据分流的规则和数据的特点，将数据流往不同的数据管道输出；

Duplicator：将数据流复制后在多个数据管道里重复执行；

Sorter：对数据进行排序；

Aggregator：对数据进行累计运算，例如Sum、Count、Max、Min等聚合运算；

Converter：对单个字段进行基本的转换，如值映射、字段拆分、空值处理、数据规范化等。

Function：提供了大量的函数进行字段值的运算，如SQRT(A)、Date A+B days、Year of Date A、Month of Date A等，且开发人员可以根据需要自行设计、添加所需的函数。

数据加载：

√TableOutput：将数据通过sql语句加载到目标数据库管理系统中；如Oracle、Mysql、SQLServer等。

√DBLoader：以外部文件批量装载的方式装载数据到目标数据库中，例如SybaseASE中的Bulk Copy，Sybase IQ的Load；

√FileOutput：将数据输出到文本文件中，如excel、txt和xml等。这个文本文件可以是本地的，也可以是存储在网络上其他位置的。

所述的DataStream(D)为：两个Step之间的数据、数据之间的关系、表名、字段名和一些参数，可以作如下定义：输入DataStream为，输出DataStream为，每一个Step的输入DataStream同时也是上一个Step的输出DataStream。

S2：构建与所述ETL模型工具对应的流程图T(即Transformation的过程)，流程图T包括若干个有向图G，定义为G1…Gn；

根据步骤S1中建立的基于工作流的ETL模型工具和有向图的定义G＝(V，E)。定义ETL流程图T＝(V，E)，其中V＝S∪D；如图2所示：定义ETL State为一个无环的有向图G，同时也代表了一个ETL过程，ETL state代表变换前后的状态；

在基于工作流的ETL工具里，ETL流程图的优化一直是近几年研究的热点。它能够增强ETL的容错性，提高ETL的执行效率，同时对判断ETL功能的正确性也有一定的效果，尽管目前为止，还没有一个理论或工具能够证明一个给定的ETL流程图的正确性。但作为一个正确的ETL过程，有向图G肯定是一个无环的有向图。

ETL的优化实际上就是在输入和输出不变的条件下，将一个ETLState转换为另一个ETLState，以提高ETL执行效率的过程，如：对表1中所示数据源，将所有性别为1(1表示男，0表示女)，且地址在001(001表示南京，002表示上海)的记录抽取出来然后加载到目标数据库中。

表1

图3为三种不同的流程设计示意图，很显然，流程2的执行效率要高于流程1，流程3的执行效率高于流程2，且它们的输入输出相同，分析可得流程2相比于流程1应用的是“交换”的变换方式，流程3相比于流程2应用的是“合并”的变换方式。

S3：将ETL过程执行的窗口和容错性等常见的限制条件考虑进去，在ETLState之间设置变换、分解、合并、串行化、并行化、添加还原点、分流和复制八种变换规则以优化数据转换过程；

所述交换Swap(v₁,v₂)具体为：将节点v₁和节点v₂相互交换；设交换前ETL流程图为G(V,E)，交换后为G'(V',E')，显然V'＝V，令E'＝E，对所有的e'∈E'，且e'＝(v,v₁)，v∈V，都变为e'＝(v,v₂)；同理，对所有的e'∈E'，且e＝(v₂,v)，v∈V，都变为e'＝(v₁,v)；再将e'＝(v₁,v₂)，变为e'＝(v₂,v₁)，具体见图4。

所述分解Split(v₁₊₂,v₁,v₂)具体为：将节点v₁₊₂分解为v₁、v₂；设分解前ETL流程图为G(V,E)，分解后为G'(V',E')，则V'＝V∪{v₁+v₂}-{v₁₊₂}；令E'＝E，对所有的e'∈E'且e'＝(v,v₁₊₂)，v∈V，都变为e'＝(v,v₁)，对所有的e'∈E'且e'＝(v₁₊₂,v)，v∈V，都变为e'＝(v₂,v)，再加上边e'＝(v₁,v₂)，具体见图5。

所述合并Merge(v₁,v₂,v₁₊₂)具体为：将节点v₁、v₂合并为v₁₊₂；设合并前ETL流程图为G(V,E)，合并后为G'(V',E')，则V'＝V-{v₁+v₂}∪{v₁₊₂}；令E'＝E，对所有的e'∈E'且e'＝(v,v₁)，v∈V，都变为e'＝(v,v₁₊₂)，对所有的e'∈E'且e'＝(v₂,v)，v∈V，都变为e'＝(v₁₊₂,v)，再去掉边e'＝(v₁,v₂)；具体见图5；

所述串行化Factorize(v₁,v₂,v₃,v₁₊₂)具体为：将处于不同的数据流中，但是类型相同的节点v₁、v₂合并为一个新的节点v₁₊₂；分流和合流可以算是并行化和串行化里的一种特殊情况；设串行化前ETL流程图为G(V,E)，串行化后为G'(V',E')。则V'＝V-{v₁+v₂}∪{v₁₊₂}。令E'＝E，对所有的e'∈E'且e'＝(v,v₁)或e'＝(v,v₂)，v∈V，都变为e'＝(v,v₃)。对所有的e'∈E且e'＝(v₃,v)，v∈V，都变为e'＝(v₁₊₂,v)，再去掉边e'＝(v₁,v₃)、边e'＝(v₁,v₃)，加上边e'＝(v₃,v₁₊₂)；具体见图6。

所述并行化Distribute(v₁,v₂,v₃,v₁₊₂)具体为：将节点v₁₊₂分解为两个类型相同的节点v₁、v₂，让节点v₁、v₂处于不同的数据流中；设并行化前ETL过程图为G(V,E)，并行化后为G'(V',E')。则V'＝V∪{v₁+v₂}-{v₁₊₂}。令E'＝E，对所有的e'∈E'且e'＝(v,v₃)，v∈V，变为e'＝(v,v₁)或e'＝(v,v₂)。对所有的e'∈E且e'＝(v₁₊₂,v)都变为e'＝(v₃,v)，再加上边e'＝(v₁,v₃)、边e'＝(v₁,v₃)，去掉边e'＝(v₃,v₁₊₂)，具体见图6。

所述添加还原点AddRecoveryPoint(v,v_RP)具体为：在节点v处添加还原点v_RP；设添加还原点前ETL流程图为G(V,E)，添加还原点后为G'(V',E')。则V'＝V∪v_RP。令E'＝E，对所有的e'∈E'且e'＝(v₁,v)，v₁∈V，添加e'＝(v₁,v_RP)，具体见图7。

所述分流Partion(v₁,v₂,n,P)具体为：将节点v₁、v₂之间的数据流分为n份；设分流前ETL流程图为G(V,E)，分流后为G'(V',E')，则V'＝V∪{v_s,v_m}；令E'＝E，对所有的e'∈E'且e'＝(v₁,v)，v∈V，变为e'＝(v₁,v_s)和e'＝(v_s,v)。对所有的e'∈E'且e'＝(v,v₂)，v∈V，变为e'＝(v,v_s)和e'＝(v_s,v₂)，具体见图8。

所述复制Replicate(v₁,v₂,n)：将节点v₁、v₂之间的数据复制n份，设复制前ETL过程图为G(V,E)，复制后为G'(V',E')。则V'＝V∪{v_s,v_m}。令E'＝E，对所有的e'∈E'且e'＝(v₁,v)，v∈V，变为e'＝(v₁,v_s)和e'＝(v_s,v)。对所有的e'∈E'且e'＝(v,v₂)，v∈V，变为e'＝(v,v_s)和e'＝(v_s,v₂)，具体见图9。

S4；在ETL工具整个流程执行过程中，利用所述八种变换规则来改变有向图G，直至选出变换后时间最小的所有有向图G，以实现优化整个ETL过程。

所述步骤S4具体为：使用穷举法将所述的八种变换规则都代入所述过程模型的ETL流程对应的有向图T，然后比较，选出变换后时间最小的有向图。

所述比较、选出变换后时间最小的有向图的具体过程为：使用open和close两个状态队列，来存储未访问的和访问过的状态，利用S_MIN来表示时间最小的状态，初始化时，让S_MIN等于初始状态S_G，open包含S_G，close为空，对open队列里的所有状态S，首先将S出队列，然后S与S_MIN进行比较，如果其时间小于S_MIN，那么S为新的最小时间状态，再让S根据所述的八种变换规则产生新的状态S'，如果S'既不属于open也不属于close，则将其送入队列open，然后将S送入close队列里，标记为已访问；不断重复上述步骤，直到open队列为空，从而实现优化整个ETL过程。

在进行有向图的变换时，遵循以下规则：根据组成Transformation节点类型的不同，将Transformation总体上分为四类：只包括Swap、Split、Merge、Factorize和Distribute的普通Step组成的Transformation、包含分流节点的Transformation、包含还原点的Transformation、包含复制节点的Transformation；

(1)对于普通Transformation，其包括Swap、Split、Merge、Factorize和Distribute，只做相邻有向图之间的变换；

(2)对于包含分流节点的Transformation，在有并行化需求时，将DataStream分开在不同的物理处理器上处理，也就是所谓的分流或者并行处理；

(3)对于包含还原点的Transformation，当发生错误时，将当前的阻塞队列里的数据全部拷贝到硬盘里，将距离错误位置最近的还原点的数据读入程序，并且从这个还原点开始执行；由于并没有执行还原点之前的流程，所以从还原点开始执行要比重新执行整个流程要快；

(4)对于包含复制节点的Transformation，当对ETL流程的容错性要求高，而对效率的要求宽松时，采用复制ETL数据流，并将其多次执行的方法来用执行效率换取容错性。

以上显示和描述了本发明的基本原理和主要特征和本发明的优点。本行业的技术人员应该了解，本发明不受上述实施例的限制，上述实施例和说明书中描述的只是说明本发明的原理，在不脱离本发明精神和范围的前提下，本发明还会有各种变化和改进，这些变化和改进都落入要求保护的本发明范围内。本发明要求保护范围由所附的权利要求书及其等效物界定。