KaiwuDBKaiwuDB

搜索
免费试用
加载中~
收藏

数据集成时表模型同步方法解析

KaiwuDB 社区官方号 time2024-08-15

线上沙龙 - 技术流第 39 期回放来啦





数据治理的第一步,也是数据中台的一个基础功能 — 即将来自各类业务数据源的数据,同步集成至中台 ODS 层。

数据湖通常是 OLAP 型的数据库,虽然中台系统很少会采用多种数据仓库技术,但在技术选型时,可选项也有很多。数据集成可实现多种异构数据库间的数据同步,想要提升数据中台的易用高效性,自动表模型同步则显得至关重要。


02

数据平台整体架构


整体来看数据平台共有 3 层:业务数据源、数据平台中的数据湖仓及数据应用。

数据集成时表模型同步方法解析(图1)


数据源可包含:
  • 关系型或非关系型数据库;

  • 某个消息系统如 MQTT、Kafka 或 RabbitMQ;

  • 放置在文件服务器上的 CSV 、XML 或 JSON 文件;

  • 非结构化日志或是类似通过 syslog 这样的协议提供的数据;

  • HTTP 接口,通常是 Restful API。


数据湖仓通过一系列数据处理,支撑数据应用。数据应用则包括数据 BI 分析工具、即席查询、开放 API、数据门户等功能板块。

其中,数据湖通常不是通过一步处理即可得到所需数据,所以数据湖也存在分层。比较常见的分层类别:ODS、DWD、DWS 和 ADS 层。有时也会将维度表的数据划分至 DIM 层,不同数据平台系统的分层会有所增减。

本期分享关注重点聚焦在 ODS 层。来自各数据源的数据会通过相应的数据集成技术,几乎不经过处理,直接存放至 ODS 层;数据平台对 ODS 层数据再经过多次处理即可得到数据应用所需数据。

  • ODS: Operational Data Store,运营数据层,贴源层

  • DWD: Data Warehouse Details,数据细节层

  • DWS: Data Warehouse Service,数据服务层

  • ADS: Application Data Service,应用数据层

  • DIM: Dimension Data Layer,维度数据层




03

各类数据源集成方式


3.1 数据库数据源的数据集成方式

数据库数据源有两种集成方式:批量同步和实时同步。

批量同步通过定时执行任务来集成数据。几乎所有的数据库都实现了 JDBC 接口,可借助数据库提供的 JDBC 驱动,通过全量或增量方式从这个数据库中获取数据再导入到数据仓库 ODS 层。

数据集成时表模型同步方法解析(图2)


实时同步批量同步有一个定时任务,执行完该同步任务即结束;而实时同步方式的同步任务是一直在执行,一旦数据库数据源的数据发生变化,它立刻即能获取到该数据变化情况并同步至 ODS 层。

数据集成时表模型同步方法解析(图3)


3.2 非数据库数据源的数据集成方式

这里主要介绍文件源、API 数据源、消息源和日志源。

文件源:有些业务系统的数据库可能不会开放给数据平台直接连接,而是将数据定时导出为 CSV 文件或类似文件,存放到 SFTP 或者是 S3 等文件服务器。数据平台集成引擎可定时从文件服务器上获取文件,解析数据并导入到 ODS 层。

数据集成时表模型同步方法解析(图4)


API 数据源:通常是一个 Restful 接口,调用该接口即能获取到结构化/半结构化数据,比如 JSON、XML 等格式。数据集成引擎作为 HTTP 的客户端,定时调用 API 接口获取数据,并转换成数据记录,通过 JDBC 方式写入到数据湖。

数据集成时表模型同步方法解析(图5)


消息源:部分业务系统会选择将数据写到消息服务器中,比如 MQTT、AMQP、 Kafka 等。数据中台集成引擎作为消息的“消费者”,负责连接到消息服务器、监听消息队列或 Topic 中的消息,获取来自业务系统的消息并将其转换成数据记录,最后通过 JDBC 方式写入到数据湖。


数据集成时表模型同步方法解析(图6)


日志源:部分业务系统看、可通过 syslog 协议方式将日志写入 syslog 服务端。数据引擎此时可以被视作服务器端,在一个端口监听客户端请求。业务系统端作为 syslog 的客户端,将日志写入到数据引擎监听的端口,数据平台集成引擎收到数据后可解析并转换成数据记录,再通过 JDBC 方式写入到数据湖。

数据集成时表模型同步方法解析(图7)



04

数据集成时模型的创建和同步


4.1 数据库数据源的模型映射

  • 来自数据库的数据源,其数据模型可以直接映射;

  • 字段名可以从源数据库获取;

  • 字段类型先通过 JDBC 驱动获取到 JDBC 的数据类型,再转换为数仓库的数据类型(这里以 PostgreSQL 为例);

  • 从 JDBC 数据类型到数据库的数据类型映射是可以一对多的,可以由自定义逻辑决定;

数据集成时表模型同步方法解析(图8)


  • 可将获取到的 JDBC 数据类型先转换为自定义的中间数据类型,然后映射为数仓数据库的数据类型。


数据集成时表模型同步方法解析(图9)

数据类型映射表:


  • 源库数据类型对应的 JDBC 数据类型由数据库对应的 JDBC 决定;

  • JDBC 数据类型在 JDK 中的 java.sql.Types 类中定义;

  • JDBC 数据类型对应的数仓库数据类型集成引擎决定。


数据集成时表模型同步方法解析(图10)


4.2 非数据库数据源的模型映射


  • 对于数据源本身不具备 schema 信息的,无法自动映射;

  • 先手动定义模型元数据,再根据模型元数据在数据中创建表。


源数据类型可以是字符型、整形或日期型。数据仓库以 PostgreSQL 为例,字符串类型对应类型包括:text 、char、varchar 等。布尔类型在数仓中可以用 boolean 、varchar 或 bit 类型。整形可以是 int4、int8、int16。数据模型的确定取决于数据平台用户如何创建模型。


数据集成时表模型同步方法解析(图11)


4.3 数据库数据源的数据模型的同步


举例说明,数据源是 MySQL 数据库,其中有一个 nation 表,包括了四个字段,字段类型有:int、char、varchar。此处示例表中的一条数据。

数据集成时表模型同步方法解析(图12)


情景1:那么该数据模型如何同步至数据中台呢?


在数据集成时,数据集成引擎读取源表的数据,并在生成的每一条同步数据里面均包含模型信息。左侧为源数据库中的数据,它在生成的同步数据时将输出如右侧的 JSON 数据。它本身含有模型信息,包括表名、所有字段及字段类型。


图中采用的是中间数据类型,比如源库中 nationkey 字段是 int 类型,对应的中间类型是 int32;name 字段在源库中是 char(25) 类型,对应中间数据类型是 string 外加一个表示长度的 size 参数。该条同步数据也包含数据本身信息,在 payload 字段中。


数据平台集成引擎可以从同步数据中提取到源表模型信息,用于在数据仓库中创建该表的模型信息。数据平台集成引擎可根据该模型生成一个在数据库创建表的 DDL 语句。注意这里表名需添加数据源的前缀标识,因为 ODS 层可能有来自不同数据源的数据,表名可能重复。因此,在 ODS 层中创建表的时需加前缀来避免重复,同时还添加了部分控制字段。

从数据中提取模型信息示例:


数据集成时表模型同步方法解析(图13)


生成建表语句示例:


数据集成时表模型同步方法解析(图14)



情景2:在后续数据源模型发生变化时如何处理呢?


可将数据中的模型信息与数据中台中的模型进行对比,如果模型增加了字段,会生成一个表变更语句。表字段类型变化,可根据用户设定的策略做表变更或者保持不变,取决于用户的同步策略。如果字段删除了通常是不会删除的。


在获取到的数据写入 ODS 之前,先执行这个建表语句或是表变更语句。数据仓库 ODS 层的模型同步完成以后,再把需要同步的数据写入到数仓的表里面。


4.4 非结构化数据源的数据模型的同步


这里以 API 接口获取到的数据为例:

数据集成时表模型同步方法解析(图15)


当获取到类似左边这样的数据,我们无法直接从数据获取到它的数据类型信息。此时只能由数据平台的使用者手动来创建模型。有了模型后就可根据该模型来生成一条 DDL 语句,然后根据 DDL 语句在数据中台中创建该表。


4.5 数据模型创建的时机


以 KDP(数据服务平台) 所使用的 Kafka Connect JDBC 框架为例:
  • 使用 JDBC-reader 插件读取源数据库;

  • 数据记录转换为消息写入 Kafka 的 Topic 中;

  • 使用 JDBC-Writer 插件读取 Topic 中的消息,写入数仓数据库。

数据集成时表模型同步方法解析(图16)


在 Writer 端解析数据模型,进行数仓中表的创建和变更:
  • Writer 端获取到一条数据后,从数据中解析出模型信息;

  • 将模型和数仓库中的表结构比对;

  • 如果不存在,则创建;如果不一致则修改。

数据集成时表模型同步方法解析(图17)


4.6 KDP 中的数据集成架构


KDP 使用 Kafka Connect 框架实现了数据库源、API  源、文件源、日志源的数据集成:
  • 使用各种 Kafka Connect 组件对接各种数据源,将获取的数据转换为统一  SourceRecord;

  • ODS 层使用的是基于 Citus 插件的分布式 PostgreSQL,所以 sink 端可以使用 Kafka Connect JDBC 从 Kafka Topic 消费数据,写入 ODS 层;

  • 整个集成任务流由 SpringCloud DataFlow 调度执行。


数据集成时表模型同步方法解析(图18)


KDP 使用 Debezium Connector 实现基于 CDC 的数据库源实时同步
  • Debezium 提供的连接器对接各支持 CDC 的数据库;

  • Debezium 本身是基于 Kafka Connect,将 CDC 事件转换为 SourceRecord 写入 Kafka Topic;

  • Sink 端同样使用 Kafka Connect JDBC 消费数据,写入 ODS 层。


数据集成时表模型同步方法解析(图19)


KDP 支持调用大数据组件如 Flink、Spark、Sqoop2 等,使用外部计算资源实现海量数据的同步:
  • 基于 SpringCloud Dataflow 框架;

  • 各 Task 作为大数据组件客户端提交任务到外部集群;

  • 外部集群执行数据集成任务。

数据集成时表模型同步方法解析(图20)



05

数据集成时常见问题及方案


数据集成时表模型同步方法解析(图21)

数据集成时表模型同步方法解析(图22)

数据集成时表模型同步方法解析(图23)





Hi,访客请登录!欢迎分享,一起交流~

发表
暂时还没评论,等你发挥!

体验全新的分布式多模数据库解决方案

企业版 社区版

KaiwuDB 是浪潮控股的数据库企业,面向工业物联网、数字能源、车联网、智慧产业等行业领域,提供稳定安全、高性能、易运维的创新数据软件与服务。

关注官方微信

友情链接:浪潮  

© 上海沄熹科技有限公司 Shanghai Yunxi Technology Co., Ltd.    沪ICP备2023002175号-1    网站服务协议   |   隐私政策