CN115033646B - 一种基于Flink&Doris构建实时数仓系统的方法 - Google Patents

Abstract

本发明提供了一种基于Flink&Doris构建实时数仓系统的方法，包括以下步骤：S1，抽取不同业务数据库获取日志文件的变更数据，并基于FlinkCDC组件将日志文件的增量变化实时同步至Kafka消息队列，并根据FlinkCDC组件同步增量变化的业务场景，创建所需的Kafka Topic，并将业务数据库中原始的操作日志报文同步至Kafka Topic中；S2，解析Kafka Topic中的业务数据库的操作日志报文得到数据库源表，并将所述业务数据库的操作日志报文发送至实时数仓ODS层，便于通过FlinkSQL将数据库源表映射成ODS层流式源表；能快速高效的进行实时需求开发，将数据库的OLAP业务与OLTP分离，消除OLAP对OLTP性能影响，极大减轻了数据库的负载，提升了实时数据可用性和业务需求的响应速度。

Description

一种基于Flink&Doris构建实时数仓系统的方法

技术领域

本发明涉及数据库技术领域，更具体地说，涉及一种基于Flink&Doris构建实时数仓系统的方法。

背景技术

随着大数据时代的到来,当前各大公司的业务需求和内部决策对于数据实时性的要求越来越迫切，需要实时数据分析的能力来赋能。传统离线数仓的数据时效性是T+1，调度频率以天为单位，无法支撑实时场景的数据需求。即使能将调度频率设置成小时，也只能解决部分时效性要求不高的场景，对于实效性要求很高的场景还是无法支撑；公司目前部分实时OLAP分析需求都依赖于Oracle/Mysql集群的计算性能，抢占了数据库的OLTP资源，该模式下的数据库性能已出现告警，因此实时数据服务问题必须得到有效解决。

现阶段实时场景的需求是通过实时计算框架SparkStreaming、Flink进行代码开发提供实时数据服务, 当前模式对开发人员的技术要求较高,且开发周期冗长,对业务的需求响应较慢，效率低下。且实时数据建设没有规范，数据可用性较差，无法形成数仓体系，造成资源大量浪费。因此研发更快速、更简便的实时场景解决方案，实现OLAP和OLTP场景分离，提升实时业务需求的响应速度和效率是数据仓库当前急需解决的问题。

发明内容

本发明要解决的技术问题在于，针对上述技术方案存在的不足，提供一种解决实时场景开发周期冗长，数据可用性低下，实现OLAP和OLTP场景分离的基于Flink&Doris构建实时数仓系统的方法。

本发明提供一种基于Flink&Doris构建实时数仓系统的方法，所述方法包括以下步骤：

S1，抽取不同业务数据库获取日志文件的变更数据，并基于FlinkCDC组件将日志文件的增量变化实时同步至Kafka消息队列，并根据FlinkCDC组件同步增量变化的业务场景，创建所需的Kafka Topic，并将业务数据库中原始的操作日志报文同步至Kafka Topic中；

S2，解析Kafka Topic中的业务数据库的操作日志报文得到数据库源表，并将所述业务数据库的操作日志报文发送至实时数仓ODS层，便于通过FlinkSQL将数据库源表映射成ODS层流式源表；

S3，通过FlinkSQL对所述实时数仓ODS层中的Kafka数据进行处理,将实时数仓建设为DWD明细层、DIM公共维度层和DM汇总层。

在本发明所述的基于Flink&Doris构建实时数仓系统的方法中，在所述步骤S1中，根据不同业务数据库选取不同的FlinkCDC组件，其中所述FlinkCDC组件支持的业务数据库包括MySQL、Oracle、SQLServer、MongoDB、OceanBase、Postgres与TiDB。

在本发明所述的基于Flink&Doris构建实时数仓系统的方法中，所述原始的操作日志报文包括所述数据库源表的DDL语句或DML语句。

在本发明所述的基于Flink&Doris构建实时数仓系统的方法中，所述步骤S2还包括以下步骤：

S21，根据实时数仓层级建设并按照数据库源表创建ODS层所需的Kafka Topic。

在本发明所述的基于Flink&Doris构建实时数仓系统的方法中，所述步骤S2还包括以下步骤：

S22，编写解析所述数据库的操作日志报文的Flink流式程序，并将操作日志报文转换成Json数据，将数据库源表中的UPDATE更新和DELETE删除操作分别转换为INSERT插入，并把解析出来的Json数据发送到ODS层的Kafka Topic中。

在本发明所述的基于Flink&Doris构建实时数仓系统的方法中，所述步骤S3中对所述实时数仓ODS层中的Kafka数据进行处理的方式包括数据转化、维度关联与指标统计。

在本发明所述的基于Flink&Doris构建实时数仓系统的方法中，在所述步骤S3中，所述DWD明细层的数据来源于ODS层的Kafka Topic，通过大数据架构提供的FlinkSQL完成数据转化、维度关联；基于每个具体业务数据库的业务过程特点构建最细粒度的DWD明细层事实表，其中DWD明细层的数据存储在Kafka Topic消息队列中，同时也将DWD明细层的数据实时写入Doris数据库中，供明细数据查询和作为简单汇总数据的加工数据源。

在本发明所述的基于Flink&Doris构建实时数仓系统的方法中，在所述步骤S3中，所述DIM层主要用于存放各个业务主题的维度数据。

在本发明所述的基于Flink&Doris构建实时数仓系统的方法中，在所述步骤S3中，通过FlinkSQL对DWD层的Kafka Topic数据进行多维的主题汇总，其中DM层的数据存放于Doris，所述DWD层可直接对外提供实时数据服务，同时也可通过接口对外提供数据查询。

根据本发明的另一方面，还提供一种计算机设备，所述计算机设备包括：

一个或多个处理器；

存储装置，用于存储一个或多个程序，

当所述一个或多个程序被所述一个或多个处理器执行，使得所述一个或多个处理器实现如本发明任一实施例所述的基于Flink&Doris构建实时数仓系统的方法。

本发明的基于Flink&Doris构建实时数仓系统的方法能快速高效的进行实时需求开发，将数据库的OLAP业务与OLTP分离，消除OLAP对OLTP性能影响，极大减轻了数据库的负载，提升了实时数据可用性和业务需求的响应速度。

附图说明

图1是本发明一种基于Flink&Doris构建实时数仓系统的方法的流程示意图。

具体实施方式

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

需要说明的是，本发明的说明书和权利要求书及上述附图中的术语“第一”、“第二”等是用于区别类似的对象，而不必用于描述特定的顺序或先后次序。应该理解这样使用的数据在适当情况下可以互换，以便这里描述的本发明的实施例能够以除了在这里图示或描述的那些以外的顺序实施。此外，术语“包括”和“具有”以及他们的任何变形，意图在于覆盖不排他的包含，例如，包含了一系列步骤或单元的过程、方法、系统、产品或设备不必限于清楚地列出的那些步骤或单元，而是可包括没有清楚地列出的或对于这些过程、方法、产品或设备固有的其它步骤或单元。

如图1所示，提供一种基于Flink&Doris构建实时数仓系统的方法，该方法包括以下步骤：

S1，抽取不同业务数据库获取日志文件的变更数据，并基于FlinkCDC组件将日志文件的增量变化实时同步至Kafka消息队列，并根据FlinkCDC组件同步增量变化的业务场景，创建所需的Kafka Topic，并将业务数据库中原始的操作日志报文同步至Kafka Topic中；

FlinkCDC 组件的工作原理是通过抽取各个业务数据库获取日志文件的变更数据，能够捕获到数据库的所有数据变化，无需像查询同步技术（Sqoop）进行全表扫描，不增加业务数据库的负载，而且FlinkCDC组件是实时同步解析业务数据库的日志文件，数据延迟基本上是秒级。

S2，解析Kafka Topic中的业务数据库的操作日志报文得到数据库源表，并将业务数据库的操作日志报文发送至实时数仓ODS层，便于通过FlinkSQL将数据库源表映射成ODS层流式源表；

S3，通过FlinkSQL对实时数仓ODS层中的Kafka数据进行处理,将实时数仓建设为DWD明细层、DIM公共维度层和DM汇总层。

在一实施例中，在步骤S1中，根据不同业务数据库选取不同的FlinkCDC组件，其中FlinkCDC组件支持的业务数据库包括MySQL、Oracle、SQLServer、MongoDB、OceanBase、Postgres与TiDB等。

具体的，原始的操作日志报文包括数据库源表的DDL语句或DML语句。

在一实施例中，步骤S2还包括以下步骤：

S21，根据实时数仓层级建设并按照数据库源表创建ODS层所需的Kafka Topic。

在一实施例中，步骤S2还包括以下步骤：

S22，编写解析数据库的操作日志报文的Flink流式程序，并将操作日志报文转换成Json数据，将数据库源表中的UPDATE更新和DELETE删除操作分别转换为INSERT插入，并把解析出来的Json数据发送到ODS层的Kafka Topic中，目的是为数据的后续层级迁移提供流式环境。

在一实施例中，步骤S3中对实时数仓ODS层中的Kafka数据进行处理的方式包括数据转化、维度关联与指标统计等。

在一实施例中，在步骤S3中，DWD明细层的数据来源于ODS 层的Kafka Topic，通过大数据架构提供的FlinkSQL完成数据转化、维度关联等；基于每个具体业务数据的业务过程特点构建最细粒度的DWD明细层事实表，其中DWD明细层的数据存储在Kafka Topic消息队列中，同时也将DWD明细层的数据实时写入Doris数据库中，供明细数据查询和作为简单汇总数据的加工数据源。

在一实施例中，在步骤S3中，DIM层主要用于存放各个业务数据库中业务主题的维度数据，且DIM公共维度层的存储引擎必须要支持FlinkSQL的访问。一般对于维表数据比较少的情况可以使用 MySQL；对于单条数据大小比较小，查询 QPS 比较高的情况，可以使用Redis存储，降低Flink集群内存资源占用；对于数据量比较大，维表数据变化不是特别敏感的场景，可以使用 HBase 存储。

在一实施例中，在步骤S3中，通过FlinkSQL对DWD层的Kafka Topic数据进行多维的主题汇总，其中DM层的数据存放于Doris，DWD层可直接对外提供实时数据服务，同时也可通过接口对外提供数据查询。因为实时数仓本身是面向主题的，可能每个主题会关心的维度都不一样，所以需要在不同的主题下，按照这个主题关心的维度对数据进行汇总。

本实施例的另一方面，还提供一种计算机设备，计算机设备包括：

一个或多个处理器；

存储装置，用于存储一个或多个程序，

当一个或多个程序被一个或多个处理器执行，使得一个或多个处理器实现实现如本发明任意实施例所提供的基于Flink&Doris构建实时数仓系统的方法。

选用Oracle业务数据库处理为例，其中其他业务数据库处理的过程也和Oracle业务数据库处理过程相同。

具体的过程如下：

在Oracle业务数据库的订单表（t_order）中插入一条数据:insert intotableMSP.t_order(ID,VIN,CREATED_DATE) value('2021112903','LGBC1AExxxxxx','2022-05-01');

FlinkCDC会实时捕获到这条数据的插入操作日志报文：

{

"before":{},

"after":{

"ID":"2021112903",

"VIN":"LGBC1AExxxxxx",

"CREATED_DATE":"2022-05-01"

},

"source":{

"name":"MSP",

"table":"t_order",

"txId":"tx-3"

},

"op":"i",

"ts_ms":1437480416795

}

（2）FlinkCDC将这条数据发送到Kafka主题 Topic_A

（3）通过Flink程序去Kafka 主题 Topic_A 将这条日志报文解析成原本的数据库数据

{

"ID":"2021112903",

"VIN":"LGBC1AExxxxxx",

"CREATED_DATE":"2022-05-01",

"DEL":"0"

}

(4)将解析的结果数据发送到实时数仓ODS层即Kafka主题 Topic_ods_order。

(5)通过FlinkSQL 进行数据拉宽处理。Topic_ods_order 与Topic_ods_user 进行join匹配, 数据增加了一个用户电话信息，得到结果数据：

{

"ID":"2021112903",

"VIN":"LGBC1AExxxxxx",

"USER_PHONE":"188xxxxxxx",

"CREATED_DATE":"2022-05-01",

"DEL":"0"

}

将结果数据发送至实时数仓DWD层即Kafka 主题 Topic_dwd_order。

（6）使用FlinkSQL对实时数仓DWD层的数据进行统计用户当日的用户订单数：

{

"ID":"2021112903",

"VIN":"LGBC1AExxxxxx",

"USER_PHONE":"188xxxxxxx",

"ORDER_COUNT":"1",

"CREATED_DATE":"2022-05-01",

"DEL":"0"

}

将统计结果存入到实时数仓DM层Doris数据库中，供可视化工具查询。

本发明实施例提供的一种基于Flink&Doris构建实时数仓系统的方法的有益效果至少在于：

1、基于FlinkCDC日志同步技术，可以实时捕获到数据库的增量变化且不会增加数据库负载，改变了传统数仓T+1离线抽取数据的模式, 为实时OLAP分析提供了数据基础。

2、数据服务端引入Doris存储引擎，可以支持高并发查询，具有良好的实时OLAP分析性能，扩容方便，支持MySQL协议等优点。且Doris的唯一主键模式克服了传统数仓关于历史数据变化而导致统计不准确的痛点，可适用于复杂的业务场景，满足数据源的增删改操作。

3、本方案使用FlinkSQL进行实时业务场景开发，可以简化实时项目的开发流程，缩短开发周期，并且降低开发难度，能快速的响应业务需求。

需要说明的是，对于前述的各方法实施例，为了简单描述，故将其都表述为一系列的动作组合，但是本领域技术人员应该知悉，本发明并不受所描述的动作顺序的限制，因为依据本发明，某些步骤可以采用其他顺序或者同时进行。其次，本领域技术人员也应该知悉，说明书中所描述的实施例均属于优选实施例，所涉及的动作和模块并不一定是本发明所必须的。

通过以上的实施方式的描述，本领域的技术人员可以清楚地了解到根据上述实施例的方法可借助软件加必需的通用硬件平台的方式来实现，当然也可以通过硬件，但很多情况下前者是更佳的实施方式。基于这样的理解，本发明的技术方案本质上或者说对现有技术做出贡献的部分可以以软件产品的形式体现出来，该计算机软件产品存储在一个存储介质(如 ROM/RAM、磁碟、光盘)中，包括若干指令用以使得一台终端设备(可以是手机，计算机，服务器，或者网络设备等)执行本发明各个实施例所述的方法。

因此，以上所述，仅为本发明较佳的具体实施方式，本发明的保护范围并不局限于此，任何熟悉本技术领域的技术人员在本发明揭露的技术范围内，可轻易想到的变化或替换，都应涵盖在本发明的保护范围之内，本发明的保护范围应该以权利要求的保护范围为准。

Claims (8)

1.一种基于Flink&Doris构建实时数仓系统的方法，其特征在于，所述方法包括以下步骤：

S1，抽取不同业务数据库获取日志文件的变更数据，并基于FlinkCDC组件将日志文件的增量变化实时同步至Kafka消息队列，并根据FlinkCDC组件同步增量变化的业务场景，创建所需的Kafka Topic，并将业务数据库中原始的操作日志报文同步至Kafka Topic中；

S2，解析Kafka Topic中业务数据库的操作日志报文得到数据库源表，并将所述业务数据库的操作日志报文发送至实时数仓ODS层，便于通过FlinkSQL将数据库源表映射成ODS层流式源表；

S3，通过FlinkSQL对所述实时数仓ODS层中的Kafka数据进行处理，将实时数仓建设为DWD明细层、DIM公共维度层和DM汇总层；

所述步骤S2还包括以下步骤：

S21，根据实时数仓层级建设并按照数据库源表创建ODS层所需的Kafka Topic；

所述步骤S2还包括以下步骤；

S22，编写解析所述业务数据库的操作日志报文的Flink流式程序，并将操作日志报文转换成Json数据格式，将数据库源表中的UPDATE更新和DELETE删除操作分别转换为INSERT插入，并把解析出来的Json数据发送到ODS层的Kafka Topic中。

2.根据权利要求1所述的基于Flink&Doris构建实时数仓系统的方法，其特征在于，在所述步骤S1中，根据不同业务数据库选取不同的FlinkCDC组件，其中所述FlinkCDC组件支持的业务数据库包括MySQL、Oracle、SQLServer、MongoDB、OceanBase、Postgres与TiDB。

3.根据权利要求2所述的基于Flink&Doris构建实时数仓系统的方法，其特征在于，所述原始的操作日志报文包括所述数据库源表的DDL语句或DML语句。

4.根据权利要求1所述的基于Flink&Doris构建实时数仓系统的方法，其特征在于，所述步骤S3中对所述实时数仓ODS层中的Kafka数据进行处理的方式包括数据转化、维度关联与指标统计。

5.根据权利要求4所述的基于Flink&Doris构建实时数仓系统的方法，其特征在于，在所述步骤S3中，所述DWD明细层的数据来源于ODS层的Kafka Topic，通过大数据架构提供的FlinkSQL完成数据转化、维度关联；基于每个具体业务数据库的业务过程特点构建最细粒度的DWD明细层事实表，其中DWD明细层的数据存储在Kafka Topic消息队列中，同时也将DWD明细层的数据实时写入Doris数据库中，供明细数据查询和作为简单汇总数据的加工数据源。

6.根据权利要求5所述的基于Flink&Doris构建实时数仓系统的方法，其特征在于，在所述步骤S3中，所述DIM公共维度 层主要用于存放各个业务数据库中业务主题的维度数据。

7.根据权利要求6所述的基于Flink&Doris构建实时数仓系统的方法，其特征在于，在所述步骤S3中，通过FlinkSQL对DWD明细 层的Kafka Topic数据进行多维的主题汇总，其中DM汇总 层的数据存放于Doris，所述DWD明细 层可直接对外提供实时数据服务，同时也可通过接口对外提供数据查询。

8.一种计算机设备，其特征在于，包括：

一个或多个处理器；

存储装置，用于存储一个或多个程序，

当所述一个或多个程序被所述一个或多个处理器执行，使得所述一个或多个处理器实现如权利要求1～7中任一所述的基于Flink&Doris构建实时数仓系统的方法。

Images