DataWorks：Hologres單表即時同步至Kafka

一、選擇同步任務類型

1. 進入Data Integration頁面。

   登入DataWorks控制台，切換至目標地區後，單擊左側導覽列的Data Integration > Data Integration，在下拉框中選擇對應工作空間後單擊進入Data Integration。

2. 在左側導覽列單擊同步任務，然後在頁面頂部單擊建立同步任務，進入同步任務的建立頁面，配置如下基本資料。

   • 資料來源和去向：Hologres→Kafka

   • 新任務名稱：自訂同步任務名稱。

   • 同步類型：單表即時。

   • 同步步驟：選擇全量同步。

二、網路與資源配置

1. 在網路與資源配置地區，選擇同步任務所使用的資源群組。您可以為該任務分配任務資源佔用CU數。

2. 來來源資料源選擇已添加的Hologres資料來源，去向資料來源選擇已添加的Kafka資料來源後，單擊測試連通性。

3. 確保來來源資料源與去向資料來源均連通成功後，單擊下一步。

三、配置同步鏈路

四、警示配置

為避免任務出錯導致業務資料同步延遲，您可以對同步任務設定警示策略。

1. 單擊頁面右上方的警示配置，進入即時子任務警示設定頁面。

2. 單擊新增警示，配置警示規則。

   說明

   此處定義的警示規則，將對該任務產生的即時同步子任務生效，您可在任務配置完成後，進入即時同步任務運行與管理介面查看並修改該即時同步子任務的監控警示規則。

3. 管理警示規則。

   對於已建立的警示規則，您可以通過警示開關控制警示規則是否開啟，同時，您可以根據警示層級將警示發送給不同的人員。

五、進階參數配置

同步任務提供部分參數可供修改，您可以按需對該參數值進行修改。

1. 單擊介面右上方的進階參數配置，進入進階參數配置頁面。

2. 在進階參數配置頁面修改相關參數值。

六、資源群組配置

您可以單擊介面右上方的資源群組配置，查看並切換當前的任務所使用的資源群組。

七、執行同步任務

1. 完成所有配置後，單擊頁面底部的完成配置。

2. 在Data Integration > 同步任務介面，找到已建立的同步任務，單擊操作列的啟動。

3. 單擊工作清單中對應任務的名稱/ID，查看任務的詳細執行過程。

查看任務運行狀態

建立完成同步任務後，您可以在同步任務頁面查看當前已建立的同步工作清單及各個同步任務的基本資料。

• 您可以在操作列啟動或停止同步任務，在更多中可以對同步任務進行編輯、查看等操作。

• 已啟動的任務您可以在執行概況中看到任務啟動並執行基本情況，也可以單擊對應的概況地區查看執行詳情。

Hologres到Kafka的單表即時同步任務分為三個步驟：

• 結構遷移：包含目標表的建立方式（已有表或自動建表），如果是自動建表，將會為您展示建表的DDL。

• 全量初始化：如果您的任務同步步驟選擇了全量同步，此處將展示全量初始化進度。

• 即時資料同步：包含即時同步的統計資訊，包含即時的讀寫流量、髒資料、Failover和作業記錄。

任務重跑

在某些特殊情況下，如果需要修改同步欄位、調整目標表欄位或表名資訊時，您還可以單擊同步任務操作列的重跑，系統會將調整的欄位、變更的目標包等資訊進行同步，之前同步過未修改的表將不會再進行同步。

• 不修改任務配置，直接單擊重跑操作，重新運行一次同步任務。

• 編輯任務，進行修改操作後，單擊完成配置。此時任務的操作會變成應用程式更新，單擊應用程式更新會直接觸發修改後的任務重跑。即時同步任務會按照新的配置運行。

Canal CDC

Canal CDC是Alibaba Canal定義的一種CDC資料格式。

• 欄位及含義

  欄位名	欄位值含義
  id	固定取值為0。
  database	Hologres資料庫名稱。
  table	Hologres表名稱。
  pkNames	Hologres表主鍵列名。
  isDdl	Binlog記錄是否對應DDL變更，由於不支援同步Hologres表DDL變更，該欄位始終取值false。
  type	DML變更類型，可選值為INSERT/UPDATE/DELETE，對應插入、更新和刪除。 說明 注意Hologres表中的一次更新會產生兩條type為UPDATE的記錄寫入Kafka Topic中 一條對應更新前的資料內容。 一條對應更新後的資料內容。 對全量同步資料，type固定取值為INSERT。
  es	13位毫秒時間戳記，對應Hologres表對應的資料變更發生時間。 在全量同步資料時，es固定取值為0。
  ts	13位毫秒時間戳記，對應Hologres BInlog記錄被同步任務讀到的時間。
  sql	Binlog記錄在上遊資料來源DDL變更時，記錄DDL變更的SQL內容，但由於不支援同步Hologres表DDL變更，該欄位始終取Null 字元串。
  sqlType	Hologres表中各欄位的資料類型對應的SQL欄位類型。 各Hologres資料類型對應sqlType取值如下： bigint：-5 scale非0的decimal：3 scale為0的decimal：-5 boolean：16 date：91 float4：6 float8：8 integer：4 smallint：5 json：12 text：12 varchar：12 timestamp：93 timestamptz：93 bigserial：-5 bytea：12 char：12 serial：4 time：92 int4[]：12 int8[]：12 float4[]：12 float8[]：12 boolean[]：12 text[]：12
  mysqlType	Hologres表各欄位的資料類型對應的MySQL欄位類型。 各Hologres資料類型對應MySQL欄位類型取值如下： bigint：BIGINT int4：INT scale非0的decimal：DECIMAL(xx,xx) scale為0的decimal：BIGINT boolean：BOOLEAN date：DATE float4：FLOAT float8：DOUBLE integer：INT smallint：SMALLINT json：TEXT text：TEXT varchar：VARCHAR(xx) timestamp：DATETIME(6) timestamptz：DATETIME(6) bigserial：BIGINT bytea：TEXT char：TEXT serial：INT time：TIME(6) int4[]：TEXT int8[]：TEXT float4[]：TEXT float8[]：TEXT boolean[]：TEXT text[]：TEXT
  data	資料變更內容，以表列名作為key，將表列變更資料內容序列化為字串後作為value組裝為JSON格式字串。具體序列化方式見下文JSON序列化說明。
  old	Hologres表中的一次更新會產生兩條type為UPDATE的記錄寫入Kafka Topic中。 變更前資料的記錄，使用old欄位記錄變更前資料內容，其他類型DML變更都使用data欄位記錄資料內容。

• Hologres Binlog INSERT DML資料對應的Canal JSON格式資料樣本

  {
      "id": 0,
      "database": "test",
      "table": "tp_int",
      "pkNames": [
          "id"
      ],
      "isDdl": false,
      "type": "INSERT",
      "es": 1640007049196,
      "ts": 1639633142960,
      "sql": "",
      "sqlType": {
          "bigint": -5,
          "integer": 4,
          "smallint": 5
      },
      "mysqlType": {
          "bigint": "BIGINT",
          "integer": "INT",
          "smallint": "SMALLINT"
      },
      "data": [
          {
              "bigint": "9223372036854775807",
              "integer": "2147483647",
              "smallint": "32767"
          }
      ],
      "old": null
  }

• Hologres全量同步資料對應的Canal JSON格式資料樣本

  {
      "id": 0,
      "database": "test",
      "table": "tp_int",
      "pkNames": [
          "id"
      ],
      "isDdl": false,
      "type": "INSERT",
      "es": 0,
      "ts": 1639633142960,
      "sql": "",
      "sqlType": {
          "bigint": -5,
          "integer": 4,
          "smallint": 5
      },
      "mysqlType": {
          "bigint": "BIGINT",
          "integer": "INT",
          "smallint": "SMALLINT"
      },
      "data": [
          {
              "bigint": "9223372036854775807",
              "integer": "2147483647",
              "smallint": "32767"
          }
      ],
      "old": null
  }

• Hologres Binlog UPDATE DML資料對應的兩條Canal JSON格式資料樣本

  //變更前資料
  {
      "id": 0,
      "database": "test",
      "table": "tp_int",
      "pkNames": [
          "id"
      ],
      "isDdl": false,
      "type": "UPDATE",
      "es": 1640007049196,
      "ts": 1639633142960,
      "sql": "",
      "sqlType": {
          "bigint": -5,
          "integer": 4,
          "smallint": 5
      },
      "mysqlType": {
          "bigint": "BIGINT",
          "integer": "INT",
          "smallint": "SMALLINT"
      },
      "old": [
          {
              "bigint": "0",
              "integer": "0",
              "smallint": "0"
          }
      ],
      "data": null
  }
  //變更後資料
  {
      "id": 0,
      "database": "test",
      "table": "tp_int",
      "pkNames": [
          "id"
      ],
      "isDdl": false,
      "type": "UPDATE",
      "es": 1640007049196,
      "ts": 1639633142960,
      "sql": "",
      "sqlType": {
          "bigint": -5,
          "integer": 4,
          "smallint": 5
      },
      "mysqlType": {
          "bigint": "BIGINT",
          "integer": "INT",
          "smallint": "SMALLINT"
      },
      "data": [
          {
              "bigint": "9223372036854775807",
              "integer": "2147483647",
              "smallint": "32767"
          }
      ],
      "old": null
  }

• Hologres Binlog DELETE DML資料對應Canal JSON格式資料樣本

  {
      "id": 0,
      "database": "test",
      "table": "tp_int",
      "pkNames": [
          "id"
      ],
      "isDdl": false,
      "type": "DELETE",
      "es": 1640007049196,
      "ts": 1639633142960,
      "sql": "",
      "sqlType": {
          "bigint": -5,
          "integer": 4,
          "smallint": 5
      },
      "mysqlType": {
          "bigint": "BIGINT",
          "integer": "INT",
          "smallint": "SMALLINT"
      },
      "data": [
          {
              "bigint": "9223372036854775807",
              "integer": "2147483647",
              "smallint": "32767"
          }
      ],
      "old": null
  }

Json

Json是將Hologres Binlog中的變更記錄，以列表名作為Key，將列表的資料內容序列化為字串後作為value，組裝為JSON格式字串寫入Kafka Topic中。