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

01 背景介绍

数据治理的第一步，也是数据中台的一个基础功能 — 即将来自各类业务数据源的数据，同步集成至中台 ODS 层。业务数据源多种多样，单单可能涉及到的主流关系型数据库就有近十种。功能更加全面的数据中台通常还具有对接非关系型数据库、消息队列、日志源、文件源等功能。

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

02 数据平台整体架构

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

数据源可包含：

• 关系型或非关系型数据库；

• 某个消息系统如 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 层。

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

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

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

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

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

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

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

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

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

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

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

• 字段类型先通过 JDBC 驱动获取到 JDBC 的数据类型，再转换为数仓库的数据类型（这里以 PostgreSQL 为例）；

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

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

数据类型映射表：

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

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

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

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

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

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

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

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

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

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

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

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

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

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

生成建表语句示例：

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

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

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

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

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

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

4.5 数据模型创建的时机

以 KDP（数据服务平台） 所使用的 Kafka Connect JDBC 框架为例：

• 使用 JDBC-reader 插件读取源数据库；

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

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

在 Writer 端解析数据模型，进行数仓中表的创建和变更：

• Writer 端获取到一条数据后，从数据中解析出模型信息；

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

• 如果不存在，则创建；如果不一致则修改。

4.6 KDP 中的数据集成架构

KDP 使用 Kafka Connect 框架实现了数据库源、API  源、文件源、日志源的数据集成:

• 使用各种 Kafka Connect 组件对接各种数据源，将获取的数据转换为统一  SourceRecord；

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

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

KDP 使用 Debezium Connector 实现基于 CDC 的数据库源实时同步：

• Debezium 提供的连接器对接各支持 CDC 的数据库；

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

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

KDP 支持调用大数据组件如 Flink、Spark、Sqoop2 等，使用外部计算资源实现海量数据的同步:

• 基于 SpringCloud Dataflow 框架；

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

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

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

1、业务库中表模型增加、删除字段、字段类型长度变更时，数据仓库中表模型的变更如何处理答：数据记录中携带表模型信息，检测到表模型不一致时，生成并执行表变更语句。

2、数仓数据库中存在多种类似的字段类型时如何选择？答：由数据中台的集成引擎的逻辑来决定选择字段类型的策略。

3、Restful API 数据源等无法自动确定源数据模型的场景如何处理？答：手动定义数据模型信息，一键生成并执行表创建/变更语句。

4、数据仓库中各层之间模型如何同步？答：除了 ODS 层外，其它层的模型都是由中台用户定义。需要手动修改模型信息，可一键生成并执行表/创建变更语句。