Hologres单表实时同步至Kafka

一、选择同步任务类型

1. 进入数据集成页面。

   登录DataWorks控制台，切换至目标地域后，单击左侧导航栏的数据集成 > 数据集成，在下拉框中选择对应工作空间后单击进入数据集成。

2. 在左侧导航栏单击同步任务，然后在页面顶部单击新建同步任务，进入同步任务的创建页面，配置如下基本信息。

   • 数据来源和去向：Hologres→Kafka

   • 新任务名称：自定义同步任务名称。

   • 同步类型：单表实时。

   • 同步步骤：选择全量同步。

二、网络与资源配置

1. 在网络与资源配置区域，选择同步任务所使用的资源组。您可以为该任务分配任务资源占用CU数。

2. 来源数据源选择已添加的Hologres数据源，去向数据源选择已添加的Kafka数据源后，单击测试连通性。

3. 确保来源数据源与去向数据源均连通成功后，单击下一步。

三、配置同步链路

四、报警配置

为避免任务出错导致业务数据同步延迟，您可以对同步任务设置报警策略。

1. 单击页面右上方的报警配置，进入实时子任务报警设置页面。

2. 单击新增报警，配置报警规则。

   说明

   此处定义的报警规则，将对该任务产生的实时同步子任务生效，您可在任务配置完成后，进入实时同步任务运行与管理界面查看并修改该实时同步子任务的监控报警规则。

3. 管理报警规则。

   对于已创建的报警规则，您可以通过报警开关控制报警规则是否开启，同时，您可以根据报警级别将报警发送给不同的人员。

五、高级参数配置

同步任务提供部分参数可供修改，您可以按需对该参数值进行修改。

1. 单击界面右上方的高级参数配置，进入高级参数配置页面。

2. 在高级参数配置页面修改相关参数值。

六、资源组配置

您可以单击界面右上方的资源组配置，查看并切换当前的任务所使用的资源组。

七、执行同步任务

1. 完成所有配置后，单击页面底部的完成配置。

2. 在数据集成 > 同步任务界面，找到已创建的同步任务，单击操作列的启动。

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变更，该字段始终取空字符串。
  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中。