第 6 章 MySQL 的 Debezium 连接器

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

当数据库客户端查询数据库时，客户端将使用数据库的当前架构。但是，数据库模式可以随时更改，这意味着连接器必须能够识别每个插入、更新或删除操作被记录的时间。另外，连接器不一定将当前的模式应用到每个事件。如果事件相对旧，则应用当前模式之前可能会记录该事件。

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

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

此数据库架构历史记录主题仅用于内部连接器。另外，连接器也可以将 模式更改事件发送到用于消费者应用程序的不同主题。

当 MySQL 连接器捕获表中的更改时，会应用 gh-ost 或 pt-online-schema-change 等模式更改，在迁移过程中会创建帮助程序表。您必须配置连接器来捕获这些帮助程序表中的更改。如果消费者不需要为帮助程序表生成的记录，请配置 单个消息转换(SMT) 从连接器发出的消息中删除这些记录。

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

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

{
  "schema": { },
  "payload": {
      "source": {  1
        "version": "2.3.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"
            }
          ]
        }
      }
    ]
  }
}

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

当 Debezium MySQL 连接器首次启动时，它会执行数据库的初始 一致快照。这个快照可让连接器为数据库的当前状态建立基准。

Debezium 可以在运行快照时使用不同的模式。快照模式由 snapshot.mode 配置属性决定。属性的默认值为 初始。您可以通过更改 snapshot.mode 属性的值来自定义连接器创建快照的方式。

您可以在以下部分找到有关快照的更多信息：

连接器在执行快照时完成一系列任务。快照模式以及对数据库有效的表锁定策略的具体步骤会有所不同。当 Debezium MySQL 连接器执行 使用全局读取锁 或 表级 锁定的初始快照时，Debezium MySQL 连接器可以完成不同的步骤。

默认情况下，连接器仅在首次启动后运行初始快照操作。在正常情况下，在这个初始快照后，连接器不会重复快照过程。连接器捕获的任何更改事件数据都只通过流处理。

然而，在某些情况下，连接器在初始快照期间获得的数据可能会过时、丢失或不完整。为了提供总结表数据的机制，Debezium 包含一个执行临时快照的选项。数据库中的以下更改可能会导致执行临时快照：

您可以通过启动所谓的 临时快照来为之前捕获的表重新运行快照。临时快照需要使用 信号表。您可以通过向 Debezium 信号表发送信号请求来发起临时快照。

当您启动现有表的临时快照时，连接器会将内容附加到表已存在的主题中。如果删除了之前存在的主题，如果启用了 自动主题创建，Debezium 可以自动创建主题。

临时快照信号指定要包含在快照中的表。快照可以捕获整个数据库的内容，或者仅捕获数据库中表的子集。另外，快照也可以捕获数据库中表的内容子集。

您可以通过将 execute-snapshot 消息发送到信号表来指定要捕获的表。将 execute-snapshot 信号类型设置为 增量，并提供快照中包含的表名称，如下表所述：

目前，execute-snapshot 操作类型仅触发 增量快照。如需更多信息，请参阅 增加快照。

为了提供管理快照的灵活性，Debezium 包含附加快照机制，称为 增量快照。增量快照依赖于 Debezium 机制 向 Debezium 连接器发送信号。

在增量快照中，除了一次捕获数据库的完整状态，就像初始快照一样，Debebe 会在一系列可配置的块中捕获每个表。您可以指定您希望快照捕获的表 以及每个块的大小。块大小决定了快照在数据库的每个获取操作期间收集的行数。增量快照的默认块大小为 1024 行。

当增量快照进行时，Debebe 使用 watermarks 跟踪其进度，维护它捕获的每个表行的记录。与标准初始快照过程相比，捕获数据的阶段方法具有以下优点：

当快照继续进行时，其他进程可能会继续访问数据库，可能会修改表记录。为了反映此类更改，INSERT、UPDATE 或 DELETE 操作会按照常常提交到事务日志。同样，持续 Debezium 流进程将继续检测这些更改事件，并将相应的更改事件记录发送到 Kafka。

对于每个数据收集，Debezium 会发出两种类型的事件，并将其存储在单个目标 Kafka 主题中。从表直接捕获的快照记录作为 READ 操作发送。同时，当用户继续更新数据收集中的记录，并且会更新事务日志来反映每个提交，Debezium 会为每个更改发出 UPDATE 或 DELETE 操作。

当快照窗口打开时，Debezium 开始处理快照块，它会向内存缓冲区提供快照记录。在快照窗口期间，缓冲区中 READ 事件的主密钥与传入流事件的主键进行比较。如果没有找到匹配项，则流化事件记录将直接发送到 Kafka。如果 Debezium 检测到匹配项，它会丢弃缓冲的 READ 事件，并将流化记录写入目标主题，因为流的事件逻辑地取代静态快照事件。在块关闭的快照窗口后，缓冲区仅包含 READ 事件，这些事件不存在相关的事务日志事件。Debezium 将这些剩余的 READ 事件发送到表的 Kafka 主题。

连接器为每个快照块重复这个过程。

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
}

Key = `test_connector`

Value = `{"type":"execute-snapshot","data": {"data-collections": ["schema1.table1", "schema1.table2"], "type": "INCREMENTAL"}}`

Key = `test_connector`

Value = `{"type":"execute-snapshot","data": {"data-collections": ["schema1.products"], "type": "INCREMENTAL", "additional-condition":"color='blue'"}}`

Key = `test_connector`

Value = `{"type":"execute-snapshot","data": {"data-collections": ["schema1.products"], "type": "INCREMENTAL", "additional-condition":"color='blue' AND brand='MyBrand'"}}`

Key = `test_connector`

Value = `{"type":"stop-snapshot","data": {"data-collections": ["schema1.table1", "schema1.table2"], "type": "INCREMENTAL"}}`

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

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

topicPrefix.databaseName.tableName

假设 fulfillment 是主题前缀，inventory 是数据库名称，数据库包含名为 orders, customers, 和 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 filename 和 binlog 位置的组合构建的。例如，如果与事务 BEGIN 事件对应的 binlog 文件名和位置分别是 mysql-bin.000002 和 1913，则 Debezium 构建的事务标识符为 file=mysql-bin.000002,pos=1913。