資料管道-同步源表結構變化

1.1 版本

1.2 歷史版本更新

1.3 應用場景

在使用管道任務進行資料實時同步程式中，來源端結構可能因業務調整等原因發生變動，如增刪表、增刪欄位、修改欄位名稱、修改欄位類型等，此時，使用者希望目標端可以自動同步這些來源端的調整，不需要人為地進行資料表的調整。

1.4 功能說明

資料管道任務支援同步源庫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 節新增欄位同理，不贅述。