图1 ROMA Connect功能视图

图2 FDI功能视图

本文将介绍FDI的关键技术-增量数据集成技术。

了解更多华为云集成技术，请关注华为云应用集成领域资深产品专家直播。

（直播链接：https://bbs.huaweicloud.com/live/DTT_live/202210271600.html）

在数据集成场景，将一个数据源中的数据定时或者实时方式同步到其他异构数据源。

数据源类型可以是API、MQ(Message queue)、DB、大数据、文件存储等。

当源端是DB，一个任务通常调度一个或几个表的部分字段数据。同时有几百上千个任务需要调度。

常见技术一：

定时方式进行数据集成任务，根据时间周期进行调度，如根据年、月、日、小时、分等为周期进行调度，每周期执行一次。每个任务可以分别设置自己的任务调度周期策略。

针对增量的数据，一般通过时间戳或者增量ID进行过滤，每次执行只同步新产生的数据，针对删除数据的同步，通过时间戳一般无法支持同步。

另外也有通过触发器记录所有变化数据到一张辅助表，定时从辅助表拉取数据进行同步。

常见技术一缺点：

机械的调度执行，不管实际有没有增量数据产生，会浪费调度资源，也会增加不必要的执行，对源数据源资源产生浪费。

调度针对每个任务调度进行调度，一般是一张数据库表一个任务，如果表的数量特别多的话，调度的负担比较大。且如果多个表之间有主外键关联关系的话，无法保证同步时实际数据的先后依赖关系，容易产生冲突导致同步失败。

采用时间戳或者增量ID同步增量数据要求表中必须有时间戳字段或者增量ID字段，用户的表不一定能满足这样的要求。

针对源表中delete的数据，通过查询方式无法获取，导致无法同步到目标数据源。

采用触发器记录所有数据的方式，由于触发器与用户的sql执行是同步执行方式，会对系统性能造成比较大的影响。存储也造成比较大的浪费。

常见技术二：

实时任务，主要面向对时延要求比较高的场景。一般调度调度一次，一直在后台执行，有增量数据产生时，可以比较及时同步到目的数据源。

针对增量数据，通常通过解析数据库的增量日志（如MySQL的binlog）来进行同步。

常见技术二缺点：

不同数据库的增量日志格式和获取接口均不同，如果需要支持多种数据库难度会比较大。

由于数据库的增量日志一般包括该数据库实例下的所有库表，不能在源端进行过滤和清洗。只能在任务执行时接收完全部数据后才能进行过滤清洗，所以会造成巨大的网络流量浪费。

• 任务调度有很多非必须的调度，即无实际增量数据的任务调度

• 非必须的数据迁移造成的网络流量浪费

• 时间戳或者增量ID对用户表结构的侵入性，无法同步delete数据。

• 对表的主外键约束关系的无法很好的支持

4.2 处理流程

本方案主要面向源端数据源是数据库类型，其他数据源类型可以根据本思路进行相应的适配。

总体步骤概述如下：

• 调度器在源端数据库中创建本发明所需要的元数据表和触发器。数据库中的数据发生变化时，通过触发器写入关键元数据到元数据表中。

• 调度器从元数据表获取所有表的变化元数据，进行依赖分析，然后调度排序

• 调度器根据策略下发任务到任务执行器进行实际任务的执行。

• 任务执行器从源端数据源中获取数据

• 任务执行器讲源端获取到的数据写入目标端数据源

4.3 元数据表结构设计

自增ID：代表元数据表的主键，主要用来在调度器中排序时使用。

表名：用来存储实际业务表的名称或者唯一标识。

Start时间戳：将所有时间进行分段，分的时间段的起始值。

End时间戳：将所有时间进行分段，分的时间段的结束值。

操作类型：用来标识对数据的操作类型，如insert、update或者对表结构的DDL操作。

主键/ddl:用来记录主键或者ddl。主键值能按范围则按照范围来存，用以节省空间。

以下针对源表是否有时间戳字段给出两张场景下的数据的示例。

1.触发器和元数据表的创建

调度器在首次连接源端数据源时，需要进行触发器和元数据表的创建。可选的也可以调度器下发任务给执行器，由执行器去执行创建任务。

4.5 调度迁移流程

1. 调度器定时访问源端数据源中的元数据表，获取关于所有表的源数据变化信息。由于是针对所有表，通常实现的定时周期一般设置比较短。根据元数据表的时间戳字段每次只取最新的未处理过的数据。针对已调度处理过的数据可以进行一定周期进行老化删除。

   注意获取数据需要严格排序，根据自增主键值进行排序，一般不能根据时间戳排序，除非时间戳可以保证前后顺序。

2. 针对同一个表的连续的多个insert/update类型操作合并，时间段取并集，如有主键也同样取并集。中间有其他类型如delete等间隔开的则不能做合并操作。

3. 针对同一个表的连续的多个delete类型操作进行合并，同样时间段取并集，主键也同样取并集。有其他类型间隔开非连续的情况下则不能做合并。ddl类型操作不做合并处理。

4. 针对上述步骤2、3中已经合并排序好的任务记录之间插入屏障，确保最终的执行先后顺序一定按排序好的执行。

5. 从源数据源获取表结构的数据，根据主外键依赖关系，将前述已经排序好的任务分配任务组。有主外键依赖分配在同一个任务组。一个任务组内根据主外键关系进行排序，针对insert/update类，主表排在子表前。针对delete则子表排在主表前。针对ddl则按原有顺序，不另外排序。

6. 调度器将排序好的任务下发到执行器进行执行。执行器执行任务时需要满足任务屏障和依赖先后关系，在此前提基础上，任务可分配在任意一个合适的执行器上执行。针对同一种任务如insert/update或delete可以根据策略适度并发执行。

7. 执行器执行任务时，根据不同场景生成不同的获取数据的sql语句。针对源表有时间戳字段insert/update,生成查询语句：select * from 表 where 时间戳 between 最早起始时间 and 最晚结束时间。无时间戳字段则生成：

   select * from 表 where 主键 in(主键值1，主键值2……) 或者 主键 between 最小主键值 and 最大主键值。

8. 执行器将查询到数据通过生成特定语句写入目标端数据源。insert/update数据生成类merge into语句，可以批量执行，当确定目标数据源中无此数据会由数据源自动做insert操作，有此数据则自动进行update。如此可以保证该任务是幂等执行，即可以重复执行，在任务执行失败时可以安全的重试。此外该语法可以将insert和update两类操作进行合并后批量执行。Oracle、SQL SERVER的语法是merge into,MySQL的类似语法是replace或 insert into on duplicate key update，PostgreSQL的语法是UPSERT，其他数据源可以参考此原理，若不支持该语句则用原生的insert或者update语法或者相对应的接口等。

9. 如果是delete类型操作，执行器则无需到源端获取数据，直接根据元数据表获取到的主键来进行同步。目标端数据源如果是数据库类型，则生成delete from 表 where 主键 in(主键值1，主键值2……) 或者 主键 between 最小主键值 and 最大主键值。如果目标端是其他类型则调用其删除接口传入主键值。