数据管道-同步源表结构变化

1.1 版本

1.2 应用场景

在使用管道任务进行数据实时同步过程中，来源端结构可能因业务调整等原因发生变动，如增删表、增删字段、修改字段名称、修改字段类型等，此时，用户希望目标端可以自动同步这些来源端的调整，不需要人为地进行数据表的调整。

1.3 功能说明

数据管道任务支持同步源库DDL功能，开启相关选项后，在源库发生DDL（删除表、新增字段、删除字段、修改字段名称、修改字段类型（长度修改 & 兼容类型修改））时，管道任务可以自动同步这些来源端变化至目标端，不需人为介入修改目标表结构。

管道任务的来源端和目标端，在下表中都为支持，才能正常使用 DDL 功能。

例如：管道任务来源端为 MySQL，目标端为 SQL Server ，支持使用 DDL 功能；来源端为 MySQL，目标端为 DB2 ，不支持使用 DDL 功能。

注：此步骤为非 SQL Server 数据库的 DDL 操作，若需要使用 SQL Server 进行 DDL 则需要对数据库进行操作，详情参见本文第四章。

示例以 MySQL 数据同步至 MySQL 数据库为例。

将 demo1 数据库中的「出入库信息」、「DEMO_PRODUCT」、「CUSTOMER」数据表实时同步至 demotest 数据库中。

CUSTOMER.xls、DEMO_PRODUCT.xls、出入库信息.xls

3.1 配置数据管道任务

3.1.1 准备工作

在 FDL 中配置管道任务前，需要准备一个独立部署的 FDL 工程、工程注册管道相关功能点、准备数据源、准备数据库环境、部署Kafka、配置传输队列、分配管道任务权限（选做）。

详情请参见：数据管道使用说明

3.1.2 选择来源

1）在有权限的文件夹下，新建管道任务。

2）在「选择来源」步骤中，选择数据源为 MySQL，同步类型选择「存量+增量同步」，同步对象选择需要同步的数据表。如下图所示：

3）点击「下一步」按钮。

3.1.3 数据去向

1）设置数据去向的数据源为 MySQL ，数据库为 demotest ，设置目标端执行物理删除，开启「同步时标记时间戳」按钮、「同步源表结构变化」按钮。如下图所示：

2）点击「下一步」按钮。

3.1.4 表字段映射

1）目标表选择自动建表，若目标数据库中已有同名表，需要修改表名；设置表字段的映射关系、为目标表设置主键。如下图所示：

2）点击「下一步」按钮。

3.1.5 管道控制

1）本文示例，该步骤默认即可，点击「保存并启动」按钮，如下图所示：

注：管道控制详细介绍请参见：配置管道任务-管道控制

2）管道任务启动后，界面如下图所示：

3.2 来源端数据变化

1）「出入库信息」表中新增了「状态」字段、修改「种类」字段为「商品种类」、删除了「期初库存」字段。如下图所示：

2）新增一条数据，如下图所示：

3.3 目标端同步结果

查看「出入库信息」目标端数据「出入库信息_1」：

• 新增字段：新增字段同步后续新增的数据（状态字段同步新增数据）。

• 删除字段：被删除字段在后续同步中传NULL值（期初库存字段同步NULL 值）。

• 修改字段名称：原名称字段在后续同步中传NULL值，同步新名称字段的后续新增数据（种类字段修改为商品种类后，种类字段停止同步，商品种类字段继续同步新增数据）。

同时在「表字段映射」中会对修改进行标记，如下图所示：

开启「同步源表结构变化」后，数据结构变化实时同步说明：

3.4 任务管理

「运行日志」界面会出现 DDL 的更改日志，如下图所示：

4.1 新增字段

对于 SQLServer 数据源，目前 DDL 同步不支持自动同步源端新增字段，若需要使用 DDL 进行新增字段的处理，请参见下文提供的方案。

4.1.1 在线方案

注：此方案如果源表在不停写入数据，源表新增字段之后，没有立刻创建新的 CDC 实例，那么在源表新增字段之后到新的 CDC 实例创建前的这段时间中，源表发生的数据修改是不会写入到新的 CDC 实例中的。这可能会导致部分数据丢失新增字段的数据，如果要完全避免这种情况，请采用离线处理方案。

1）源表执行 DDL，新增一个字段。

2）给源表新建一个 Capture Instance

使用命令：

EXEC sys.sp_cdc_enable_table
     @source_schema = N'Schema',
     @source_name = N'Table',
     @role_name = NULL, -- 如果要限制读取角色，可能需要设置
     @capture_instance = N'Schema_Table_newname'; -- 给一个新的名字

注：Table 是数据表的名称（没有 schema）。

3）等待 FineDataLink 检测到新的 Capture Instance 创建后，自动切换到新的 Capture Instance

即可看到目标表中已经同步新增的字段，如下图所示：

4）可以手动把旧的 Capture Instance 删除。

检查 change_tables 中旧的 Capture Instance 名称：

USE database;
GO

EXECUTE sys.sp_cdc_help_change_data_capture;
GO

然后禁用该 Capture Instance ，示例命令：

-- Disable a Capture Instance for a table
    EXEC sys.sp_cdc_disable_table
    @source_schema = N'Schema',
    @source_name   = N'Table',
    @capture_instance = N'Schema_Table'

注：Table 是数据表的名称(没有 schema )；Schema_Table 为需要删除的旧 Capture Instance 。

然后使用如下命令检查 change_tables 里是否有这条记录，若没有则表示已经被删除，若有的话需要手动使用 SQL 命令删除。

USE database;
GO

EXECUTE sys.sp_cdc_help_change_data_capture;
GO

4.1.2 离线处理方案

1）源表停止写入，等待 FineDataLink 同步完源表所有的数据。

2）停止管道任务。

3）源表执行 DDL，新增一个字段：

4）给源表新建一个 Capture Instance

使用命令：

EXEC sys.sp_cdc_enable_table
     @source_schema = N'Schema',
     @source_name = N'Table',
     @role_name = NULL, -- 如果要限制读取角色，可能需要设置
     @capture_instance = N'Schema_Table_newname'; -- 给一个新的名字

5）恢复源表写入。

6）启动管道任务。

7）等待 FineDataLink 检测到新的 Capture Instance 创建后，自动切换到新的 Capture Instance

即可看到目标表中已经同步新增的字段，如下图所示：

8）可以手动删除旧的 Capture Instance。

注：如果没有按照正确顺序操作，可能造成部分有新字段的数据流入了旧的 CDC 实例，那么这部分新字段的数据就丢失了。

SAP HANA 作为数据源时，程序无法感知到源表结构变化，因此需要用户手动有计划地执行表结构变更等操作，才能完成同步。

5.1 新增字段

• 正在运行的管道任务，数据表新增了字段，且需要同步新增字段。

停止源表写入，然后暂停 FineDataLink 管道任务。

进入管道任务编辑界面，移除任务中新增字段的表并保存任务，如下图所示：

在源表和目标表新增字段，如下图所示：

重新在FineDataLink 任务中添加表，选择已存在表，如下图所示：

保存并运行任务，恢复源表写入。

• 不需要同步新增字段：仅需要新增源表字段，不需要其他操作。

5.2 删除字段

停止源表写入，然后暂停 FineDataLink 管道任务。

移除任务中删除字段的表并保存任务。

在源表和目标表删除字段，重新在FineDataLink 任务中添加表，选择已存在表。

保存并运行任务，恢复源表写入。

与 5.1 节新增字段同理，不赘述。

5.3 修改字段类型或长度

停止源表写入，然后暂停 FineDataLink 管道任务。

同时在源表、目标表调整字段类型和长度，

源表修改字段类型命令实例：

ALTER TABLE "FDL_ROXY"."dingdan" ALTER ("CREATETIME" NVARCHAR (255) NULL);

启动任务，恢复源表写入。

5.4 修改字段名

停止源表写入，然后暂停 FineDataLink 管道任务。

移除任务中改字段名的表并保存任务。

在源表和目标表修改字段名。

重新在FineDataLink 任务中添加表，

保存并运行任务，恢复源表写入。

与 5.1 节新增字段同理，不赘述。