线上沙龙 - 技术流第 39 期回放来啦
02 数据平台整体架构 关系型或非关系型数据库; 某个消息系统如 MQTT、Kafka 或 RabbitMQ; 放置在文件服务器上的 CSV 、XML 或 JSON 文件; 非结构化日志或是类似通过 syslog 这样的协议提供的数据; HTTP 接口,通常是 Restful API。 ODS: Operational Data Store,运营数据层,贴源层 DWD: Data Warehouse Details,数据细节层 DWS: Data Warehouse Service,数据服务层 ADS: Application Data Service,应用数据层 DIM: Dimension Data Layer,维度数据层 03 各类数据源集成方式 消息源:部分业务系统会选择将数据写到消息服务器中,比如 MQTT、AMQP、 Kafka 等。数据中台集成引擎作为消息的“消费者”,负责连接到消息服务器、监听消息队列或 Topic 中的消息,获取来自业务系统的消息并将其转换成数据记录,最后通过 JDBC 方式写入到数据湖。 04 数据集成时模型的创建和同步 来自数据库的数据源,其数据模型可以直接映射; 字段名可以从源数据库获取; 字段类型先通过 JDBC 驱动获取到 JDBC 的数据类型,再转换为数仓库的数据类型(这里以 PostgreSQL 为例); 从 JDBC 数据类型到数据库的数据类型映射是可以一对多的,可以由自定义逻辑决定; 可将获取到的 JDBC 数据类型先转换为自定义的中间数据类型,然后映射为数仓数据库的数据类型。 数据类型映射表: 源库数据类型对应的 JDBC 数据类型由数据库对应的 JDBC 决定; JDBC 数据类型在 JDK 中的 java.sql.Types 类中定义; JDBC 数据类型对应的数仓库数据类型集成引擎决定。 对于数据源本身不具备 schema 信息的,无法自动映射; 先手动定义模型元数据,再根据模型元数据在数据中创建表。 源数据类型可以是字符型、整形或日期型。数据仓库以 PostgreSQL 为例,字符串类型对应类型包括:text 、char、varchar 等。布尔类型在数仓中可以用 boolean 、varchar 或 bit 类型。整形可以是 int4、int8、int16。数据模型的确定取决于数据平台用户如何创建模型。 举例说明,数据源是 MySQL 数据库,其中有一个 nation 表,包括了四个字段,字段类型有:int、char、varchar。此处示例表中的一条数据。 情景1:那么该数据模型如何同步至数据中台呢? 在数据集成时,数据集成引擎读取源表的数据,并在生成的每一条同步数据里面均包含模型信息。左侧为源数据库中的数据,它在生成的同步数据时将输出如右侧的 JSON 数据。它本身含有模型信息,包括表名、所有字段及字段类型。 图中采用的是中间数据类型,比如源库中 nationkey 字段是 int 类型,对应的中间类型是 int32;name 字段在源库中是 char(25) 类型,对应中间数据类型是 string 外加一个表示长度的 size 参数。该条同步数据也包含数据本身信息,在 payload 字段中。 数据平台集成引擎可以从同步数据中提取到源表模型信息,用于在数据仓库中创建该表的模型信息。数据平台集成引擎可根据该模型生成一个在数据库创建表的 DDL 语句。注意这里表名需添加数据源的前缀标识,因为 ODS 层可能有来自不同数据源的数据,表名可能重复。因此,在 ODS 层中创建表的时需加前缀来避免重复,同时还添加了部分控制字段。 情景2:在后续数据源模型发生变化时如何处理呢? 可将数据中的模型信息与数据中台中的模型进行对比,如果模型增加了字段,会生成一个表变更语句。表字段类型变化,可根据用户设定的策略做表变更或者保持不变,取决于用户的同步策略。如果字段删除了通常是不会删除的。 在获取到的数据写入 ODS 之前,先执行这个建表语句或是表变更语句。数据仓库 ODS 层的模型同步完成以后,再把需要同步的数据写入到数仓的表里面。 这里以 API 接口获取到的数据为例: 当获取到类似左边这样的数据,我们无法直接从数据获取到它的数据类型信息。此时只能由数据平台的使用者手动来创建模型。有了模型后就可根据该模型来生成一条 DDL 语句,然后根据 DDL 语句在数据中台中创建该表。 使用 JDBC-reader 插件读取源数据库; 数据记录转换为消息写入 Kafka 的 Topic 中; 使用 JDBC-Writer 插件读取 Topic 中的消息,写入数仓数据库。 Writer 端获取到一条数据后,从数据中解析出模型信息; 将模型和数仓库中的表结构比对; 如果不存在,则创建;如果不一致则修改。 使用各种 Kafka Connect 组件对接各种数据源,将获取的数据转换为统一 SourceRecord; ODS 层使用的是基于 Citus 插件的分布式 PostgreSQL,所以 sink 端可以使用 Kafka Connect JDBC 从 Kafka Topic 消费数据,写入 ODS 层; 整个集成任务流由 SpringCloud DataFlow 调度执行。 Debezium 提供的连接器对接各支持 CDC 的数据库; Debezium 本身是基于 Kafka Connect,将 CDC 事件转换为 SourceRecord 写入 Kafka Topic; Sink 端同样使用 Kafka Connect JDBC 消费数据,写入 ODS 层。 基于 SpringCloud Dataflow 框架; 各 Task 作为大数据组件客户端提交任务到外部集群; 外部集群执行数据集成任务。 05 数据集成时常见问题及方案4.2 非数据库数据源的模型映射
4.3 数据库数据源的数据模型的同步
从数据中提取模型信息示例:
生成建表语句示例:
4.4 非结构化数据源的数据模型的同步
4.5 数据模型创建的时机
以 KDP(数据服务平台) 所使用的 Kafka Connect JDBC 框架为例:
在 Writer 端解析数据模型,进行数仓中表的创建和变更:
4.6 KDP 中的数据集成架构
KDP 使用 Kafka Connect 框架实现了数据库源、API 源、文件源、日志源的数据集成:
KDP 使用 Debezium Connector 实现基于 CDC 的数据库源实时同步
KDP 支持调用大数据组件如 Flink、Spark、Sqoop2 等,使用外部计算资源实现海量数据的同步: