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

1.1 版本

1.2 应用场景

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

1.3 功能说明

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

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

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

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

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

3.1 配置数据管道任务

1）参考：配置数据管道任务 设置数据来源，选择需要同步的数据表，如下图所示：

2）同时在「选择数据去向」时，勾选「同步源表结构变化」按钮，如下图所示：

3）设置表字段的映射关系，修改目标表名并设置主键，如下图所示：

4）设置任务的脏数据阈值、勾选「失败重试」，并选择「消息通知」，勾选「源表结构变化」，并选择通知方式，如下图所示：

此步骤可保证来源端结构因业务调整等原因发生变动，如增删表、增删字段、修改字段名称、修改字段类型后用户可即时被提醒。

3.2 来源端数据变化

在设置管道任务后「出入库信息」表中新增了「状态」字段、修改「种类」字段为「商品种类」、删除了「期初库存」字段，如下图所示：

3.3 目标端同步结果

查看「出入库信息」目标端数据「in_out」，例如「序号」为11的这条数据为数据结构变化后同步的数据，在修改字段后同步的数据中：

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

• 删除字段：被删除字段在后续同步中传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 节新增字段同理，不赘述。

6.1 部分数据未完成同步

问题描述：

管道任务中，源端为 Oracle，运行一段时间后任务异常终止，报错如下：

抽数引擎内部异常-读取端出错-数据管道 Oracle LogMiner 遇到无法解析的数据，请根据日志中的提示信息操作活动联系技术支持人员。-无法正常解析表XX的数据，可能是由于该表表结构发生变化且在表结构发生变化前有未完成同步的数据，这部分未完成同步的数据无法正常处理

临时解决方案：请编辑任务将表XX移除，然后重启任务，让任务恢复正常运行，等运行日志中显示 offsetScn 大于295506865之后可以编辑任务把该表重新添加回来，重新进行全量+增量同步

解决方案：

该报错 DDL 无法解决。

需参考报错中的「临时解决方案」，解决这个问题。