基于Kafka、Disruptor技术对传统ETL的改进

王 梓，梁正和，吴莹莹

(河海大学 计算机与信息学院，江苏 南京 211110)

0 引 言

大数据时代的到来，信息已成为各行各业发展的决策依据。世界计算机数据信息总量海量增长，已经有越来越多的企业、教育、科研机构和生厂商们意识到这些数据中所包含的重要价值，ETL的发展已成为必然趋势[1]。企业在使用各种应用系统期间往往积累了大量的信息、数据资源，计算机技术的快速发展使用户可以对这些信息进行采集、分析、处理，它们成为了企业发展的决策性依据，构成了企业发展的宝贵财富。随着时间的推移，技术的进步，原有旧系统不断升级从而被新系统完全取代，在此使用期间往往会积累大量珍贵的历史数据，这些数据也是新系统成功启用的关键。同时，这些从旧系统到新系统使用过程中的历史数据也是企业进行决策发展，制定战略方向的重要依据。

如今，随着数据的大量累积，越来越多的企业、厂商都在构建数据仓库(data warehouse)来满足企业在发展过程中的战略需要，他们需要将这些可能来自不同的软硬件平台、操作系统、数据模型乃至地理上分布、管理上自治和模式上异构的数据源进行集成。一个或多个不同数据源的相关数据可以进行综合集中放入数据仓库[2]，在数据仓库中可以针对不同的主题和汇总数据进行统计和分析，从而为决策人员提供数据支持。在构建数据仓库的过程中，首先需要将各种分布的、异构的数据源中的数据抽取出来，在此过程中进行清洗、转换、集成最后加载到数据仓库中，这个过程叫做抽取转换装载即ETL(extraction transformation loading)[3]数据迁移。ETL在构建过程中需要面对传输效率、准确性、数据异构性、多目标等问题。传统的ETL在解决不同操作系统之间使用不同的编程语言问题方面，设置一个专有的转换引擎置于数据源和目标数据仓库[4]之间，用于运行所有的转换程序。但在数据转换过程中，专有引擎执行所有转换工作成为“瓶颈”。随着企业的发展，数据需要从结构化数据源(关系数据库)，非结构化数据源(PDF文件、邮件等)，半结构化数据源(XML和其他标记语言)，遗留系统(主机)、应用程序包(SAP)等异构数据源中提取，同时数据量也呈现出递增式的增长，对数据的存储[5]，数据的异构性、并发性进行研究已成为当前的主要研究方向。

在ETL构建过程中，尽量降低ETL过程的设计与维护代价，提升ETL过程的执行效率，是企业在实际项目中重要考虑的问题，因此，设计一种优秀的ETL工具对数据仓库[6]非常有益。利用ETL工具可以对异构数据源中的业务数据进行抽取和转换，并将其装载到数据仓库[7]，其主要作用是对各类业务数据的清理、转换和装载，为基于数据仓库的决策分析应用提供高质量的数据。截止目前，生产的数据量大大提高，传统数据处理和数据仓库技术已不能满足海量数据[8]处理的现实需求，因此基于Kafka和Disruptor，提出一种对传统ETL进行改进的模型[9]，并就某教育企业对全国高校系统数据迁移进行了实验研究。

1 ETL过程的特点

ETL是对数据进行抽取、清洗、转换和装载[10]的过程，数据从异构数据源中抽取，迁移到指定的目标库。其间，数据的抽取、清洗、转换和装载形成串行或并行的过程。T过程是ETL的核心,也是数据的转换,而抽取和装载一般可以作为转换的输入和输出,或者作为一个单独的部件,其复杂程度没有转换部件高。ETL是构建数据仓库的重要组成部分，用户从数据源抽取出所需的数据，经过数据清洗,最终按照预先定义好的数据仓库模型，将数据加载到数据仓库中。传统的ETL上手快，易操作，可是当数据海量增长时，传统的ETL在性能和数据处理的准确性、多样性、并发性等方面却大打折扣。现在数据的ETL过程经常会选择Kafka作为消息中间件应用在离线和实时的场景中。针对上述问题，文中基于Kafka和Disruptor并发框架对传统ETL进行了改进。

1.1 Disruptor特点

Disruptor是一个高性能的异步处理框架，一般被设计在生产者—消费者(producer-consumer problem，PCP)问题上，可以获得尽量高的吞吐量(TPS)和尽量低的延迟。针对“并发、缓冲区、生产者—消费者模型、事务处理”这些元素的程序来说，Disruptor是一种大幅提升性能(TPS)的方案。它本质上是个ringbuffer，buffer(就是数组)做过优化防止JVM伪共享，lock free是通过CAS自旋[11]，多线程[12]并发获取buffer中的序号，这里需要CAS，把事件放入槽中，工作线程调度交给jdk线程池，只要buffer中有事件，就不停提交给线程池，不需要锁进程，解决了多线程读写，实现读写同步，解决了数据延迟的问题。

(1)disruptor没有锁，所以效率高，速度快。

(2)所有访问者都记录自己序号的实现方式[13]，允许多个生产者与多个消费者共享相同的数据结构。

1.2 Kafka特点

近年来，Kafka作为一个新兴的分布式消息系统，受到了众多企业、科研机构的青睐。Kafka在分布式集群应用中作为多种类型的数据管道和消息系统[14]而应用广泛。流数据是大多数集群统计和实时数据采集过程中所产生的数据，可能包括页面访问量、物联网传感器采集数据等方式。Kafka用作LinkedIn的活动流(activity stream)和运营数据处理[15]管道(pipeline)的基础。在消息保存中Kafka根据每个topic进行归类，消息发送者称为producer，接收消息者称为consumer，此外Kafka集群[16]由多个Kafka实例组成，每个实例(server)称为broker。

(1)持久化能力的高效性。对TB级以上的数据也可以在常数时间复杂度读取和写入硬盘。

(2)支持broker间的消息分区，并保证分区中消息读取的有序性。

(3)分布式系统，易于向外扩展。所有的producer、broker和consumer都会有多个，均为分布式的。无需停机即可扩展机器。

(4)消息被处理的状态是在consumer端维护，而不是由server端维护。当失败时能自动平衡。

(5)异步：Kafka分布式[17]消息系统采用异步通信机制，消息进入系统缓存后系统无需立刻应答或处理，可以根据用户需求和配置情况选择。

(6)持久性、可靠性：消息被持久化到本地磁盘，并且支持数据备份，防止数据丢失。

2 改进后的ETL模型

基于Kafka和Disruptor数据处理技术对传统ETL的改进如图1所示。

图1 改进的ETL模型

实验中，Kafka结合Disruptor技术，数据通道在启动后，会启动一个restful数据被动接收接口，前端埋点或其他的数据收集服务会将收集到的数据以restful+json的方式传到接口，接口在接收到数据后，会根据配置的类型，将数据发送到指定的接收源，接口源可以是配置的一个read-type，是Kafka的job，实现数据之间的快速传输。从多种数据源中抽取数据，将数据发送到Disruptor的RingBuffer环形区域，Disruptor数据消费者发送数据到Kafka服务器，由于Disruptor高性能[18]、低延迟的特点，从而提高了目标源数据到Kafka中数据的速度，因此在传输速度方面对传统的ETL有很大的改进，极大节约了时间[19]。再利用Kafka高吞吐和异步的特性，Disruptor可持续发送数据到Kafka中，Kafka消费者处理数据，节省了数据等待传输时间，同时该模型可接受多种不同数据源，实现数据多样性的传输。

为此，在研发创新方面，华岳每年投入大量资金，不断提升产品质量和智能化程度，同时也收获了宝贵的自主知识产权。截至目前，公司共获得48项专利，其中发明专利7项，实用新型38项，有力支撑了产品更新升级和向高端化发展。

3 实验验证过程

启动数据通道，在通道启动一个restful数据被动接收接口，解析配置文件，程序初始化操作→启动消费者线程→生产者进程发布事件到Disruptor，从目标源中读取数据到Disruptor的RingBuffer环形队列中，Disruptor消费者将RingBuffer环形队列[20]收集到的数据通过restful+json的方式传到接口，接口在接收到数据后将数据发送到Kafka。由于broker的增加或者减少都会触发Consumer Rebalance，数据通过Kafka Consumer开始处理partition里面的message，实验中接收源为jdbc,数据发送到指定的接收源过程中对数据进行了清洗和装载。由于Kafka高吞吐、异步性的特点，可以将数据存放在Kafka服务器端，随时处理服务器端的数据。

Disruptor读取源数据和发送数据给Kafka broker服务器，在程序中的主要流图如图2所示。

图2 Disruptor读写过程

Kafka原理结构如图3所示。

图3 Kafka原理

4 实验结果分析

4.1 实验结果

图4所示为改进后的ETL将MYSQL数据库中的数据迁移到Postgresql数据库中。t_ampa_useraction分别在MYSQL不同的数据库下，在job配置文件中配置了目标源和目标数据库，启动项目，开始数据迁移。实验中使用了4张表同时进行，由开始时间和接入时间可见，极大地节约了数据传输时间。

图4 MYSQL中数据导入Postgresql

图5所示为传统ETL将MYSQL数据库中的数据迁移到Postgresql数据库中，两个ETL同时启用。

图5 传统ETL MYSQL中数据导入Postgresql

4.2 实验分析

通过实验对比，现在数据的ETL过程经常会选择Kafka作为消息中间件应用在离线和实时的场景中，结合Kakfa的特性和Disruptor高并发、高吞吐的特点，Disruptor消费者发送数据到Kafka服务器中，实现数据的高效传输。传统的ETL当数据量上升到一定程度时，传输的记录有时会缺失，处理时间过长，无法实现并发存储，对此做出了改进。在实际开发过程中，因为同时对各个高校的数据进行迁移，数据量急剧增长，因此使用Kafka高吞吐量、低延迟、异步的特性，可以极大提高数据的传输效率。

5 结束语

文中利用Kafka和Disruptor并发框架两种数据处理技术快速构建数据ETL通道，凭借高吞吐量、低延迟的特点，极大节约了数据之间的传输时间。实验采用分布式消息系统作为大规模流数据的缓存，提高了平台对动态流数据输入数据量突发性变化的适应能力。针对多种数据源如http、txt、jdbc等的处理，对传统ETL进行了改进，实现对大量数据的并发处理，使不同数据库之间的数据能够快速、同步、多样地传输。

虽然对传统ETL在处理速度和吞吐量方面进行了改进，但是在排序、分页等功能上做得还不够完善，当job上升到10个以上时，xml文件解析便容易出错，对这些问题将有待进一步完善。