第 5 章 MySQL 的 Debezium 连接器

Debezium MySQL 连接器支持以下 MySQL 拓扑：

当数据库客户端查询数据库时，客户端将使用数据库的当前模式。但是，数据库架构可以随时更改，这意味着连接器必须能够识别每次插入、更新或删除操作时的模式是什么。另外，连接器无法只使用当前的模式，因为连接器可能是处理在表模式更改前记录的事件相对旧的事件。

为确保正确处理 schema 更改后发生的更改，MySQL 仅包含对数据的行级更改，以及应用到数据库的 DDL 语句。当连接器读取 binlog 并位于这些 DDL 语句中时，它会解析它们并更新每个表架构的内存中表示。连接器使用此架构表示来标识每次插入、更新或删除操作时表的结构，并生成适当的更改事件。在单独的数据库架构历史记录 Kafka 主题中，连接器记录了所有 DDL 语句以及出现每个 DDL 语句的 binlog 中的位置。

当连接器在崩溃或安全停止后重启时，连接器会从特定位置（即从特定时间点）读取 binlog。连接器重新构建此时存在的表结构，方法是读取数据库模式历史记录 Kafka 主题，并将所有 DDL 语句解析为启动连接器的 binlog 中的点。

此数据库架构历史记录主题仅用于连接器。连接器可以选择 发出架构将事件更改为用于消费者应用程序的不同主题。

当 MySQL 连接器捕获表中的更改时，架构更改工具（如 gh-ost 或 pt-online-schema-change ）会应用在迁移过程中创建的帮助程序表。需要配置连接器来捕获对这些帮助程序表的更改。如果消费者不需要为帮助程序表生成的记录，则可以应用单个消息转换来过滤它们。

如需更多信息，请参阅接收 Debezium 事件记录 的主题的默认名称。

您可以配置 Debezium MySQL 连接器来生成模式更改事件，这些事件描述了应用于数据库中捕获的表的 schema 更改。连接器将模式更改事件改为名为 < topicPrefix> 的 Kafka 主题，其中 topicPrefix 是 topic.prefix connector 配置属性中指定的命名空间。连接器发送到 schema 更改主题的消息包含一个有效负载，也可以包含更改事件消息的 schema。

模式更改事件消息的有效负载包括以下元素：

{
  "schema": { },
  "payload": {
      "source": {  1
        "version": "2.1.4.Final",
        "connector": "mysql",
        "name": "mysql",
        "ts_ms": 1651535750218, 2
        "snapshot": "false",
        "db": "inventory",
        "sequence": null,
        "table": "customers",
        "server_id": 223344,
        "gtid": null,
        "file": "mysql-bin.000003",
        "pos": 570,
        "row": 0,
        "thread": null,
        "query": null
      },
      "databaseName": "inventory", 3
      "schemaName": null,
      "ddl": "ALTER TABLE customers ADD middle_name varchar(255) AFTER first_name", 4
      "tableChanges": [  5
        {
          "type": "ALTER", 6
          "id": "\"inventory\".\"customers\"", 7
          "table": {    8
            "defaultCharsetName": "utf8mb4",
            "primaryKeyColumnNames": [  9
              "id"
            ],
            "columns": [  10
              {
                "name": "id",
                "jdbcType": 4,
                "nativeType": null,
                "typeName": "INT",
                "typeExpression": "INT",
                "charsetName": null,
                "length": null,
                "scale": null,
                "position": 1,
                "optional": false,
                "autoIncremented": true,
                "generated": true
              },
              {
                "name": "first_name",
                "jdbcType": 12,
                "nativeType": null,
                "typeName": "VARCHAR",
                "typeExpression": "VARCHAR",
                "charsetName": "utf8mb4",
                "length": 255,
                "scale": null,
                "position": 2,
                "optional": false,
                "autoIncremented": false,
                "generated": false
              },
              {
                "name": "middle_name",
                "jdbcType": 12,
                "nativeType": null,
                "typeName": "VARCHAR",
                "typeExpression": "VARCHAR",
                "charsetName": "utf8mb4",
                "length": 255,
                "scale": null,
                "position": 3,
                "optional": true,
                "autoIncremented": false,
                "generated": false
              },
              {
                "name": "last_name",
                "jdbcType": 12,
                "nativeType": null,
                "typeName": "VARCHAR",
                "typeExpression": "VARCHAR",
                "charsetName": "utf8mb4",
                "length": 255,
                "scale": null,
                "position": 4,
                "optional": false,
                "autoIncremented": false,
                "generated": false
              },
              {
                "name": "email",
                "jdbcType": 12,
                "nativeType": null,
                "typeName": "VARCHAR",
                "typeExpression": "VARCHAR",
                "charsetName": "utf8mb4",
                "length": 255,
                "scale": null,
                "position": 5,
                "optional": false,
                "autoIncremented": false,
                "generated": false
            }
          ],
          "attributes": [ 11
            {
              "customAttribute": "attributeValue"
            }
          ]
        }
      }
    ]
  }
}

如需更多信息，请参阅 架构历史记录主题。

当 Debezium MySQL 连接器首次启动时，它会为 您的数据库执行初始一致的快照。以下流描述了连接器如何创建此快照。此流用于默认快照模式，即 。有关其他快照模式的详情，请查看 MySQL 连接器 snapshot.mode 配置属性。

SELECT * FROM <tableName> ....

SELECT * FROM <tableName> WHERE <additional-condition> ....

INSERT INTO <signalTable> (id, type, data) VALUES ('<id>', '<snapshotType>', '{"data-collections": ["<tableName>","<tableName>"],"type":"<snapshotType>","additional-condition":"<additional-condition>"}');

INSERT INTO myschema.debezium_signal (id, type, data) VALUES('ad-hoc-1', 'execute-snapshot', '{"data-collections": ["schema1.products"],"type":"incremental", "additional-condition":"color=blue"}');

INSERT INTO myschema.debezium_signal (id, type, data) VALUES('ad-hoc-1', 'execute-snapshot', '{"data-collections": ["schema1.products"],"type":"incremental", "additional-condition":"color=blue AND quantity>10"}');

{
    "before":null,
    "after": {
        "pk":"1",
        "value":"New data"
    },
    "source": {
        ...
        "snapshot":"incremental" 1
    },
    "op":"r", 2
    "ts_ms":"1620393591654",
    "transaction":null
}

默认情况下，MySQL 连接器将表中的所有 INSERT、UPDATE 和 DELETE 操作的更改事件写入特定于该表的单个 Apache Kafka 主题。

连接器使用以下惯例命名更改事件主题：

topicPrefix.databaseName.tableName

假设 fulfillment 是主题前缀，inventory 是数据库名称，数据库包含名为 orders、customer 和 products 的表。Debezium MySQL 连接器将事件发送到三个 Kafka 主题，一个用于数据库中的每个表：

fulfillment.inventory.orders
fulfillment.inventory.customers
fulfillment.inventory.products

以下列表提供默认名称组件的定义：

连接器应用类似的命名约定来标记其内部数据库架构历史记录主题、架构更改主题 以及 事务元数据主题。

如果默认主题名称不满足您的要求，您可以配置自定义主题名称。要配置自定义主题名称，您可以在逻辑主题路由 SMT 中指定正则表达式。有关使用逻辑主题路由 SMT 自定义主题命名的更多信息，请参阅 主题路由。

Debezium 为每个事务生成 BEGIN 和 END 分隔符的事务边界事件。事务边界事件包含以下字段：

{
  "status": "BEGIN",
  "id": "0e4d5dcd-a33b-11ea-80f1-02010a22a99e:10",
  "ts_ms": 1486500577125,
  "event_count": null,
  "data_collections": null
}

{
  "status": "END",
  "id": "0e4d5dcd-a33b-11ea-80f1-02010a22a99e:10",
  "ts_ms": 1486500577691,
  "event_count": 2,
  "data_collections": [
    {
      "data_collection": "s1.a",
      "event_count": 1
    },
    {
      "data_collection": "s2.a",
      "event_count": 1
    }
  ]
}

除非通过 topic.transaction 选项覆盖，否则连接器会将事务事件发送到 < topic.prefix>.transaction 主题。

以下是消息的示例：

{
  "before": null,
  "after": {
    "pk": "2",
    "aa": "1"
  },
  "source": {
...
  },
  "op": "c",
  "ts_ms": "1580390884335",
  "transaction": {
    "id": "0e4d5dcd-a33b-11ea-80f1-02010a22a99e:10",
    "total_order": "1",
    "data_collection_order": "1"
  }
}

对于没有启用 GTID 的系统，事务标识符会使用 binlog 文件名和 binlog 位置的组合来构建。例如，如果与事务 BEGIN 事件对应的 binlog 文件名和位置分别是 mysql-bin.000002 和 1913，则 Debezium 构建的事务标识符将是 file=mysql-bin.000002,pos=1913。