第 6 章 Oracle 的 Debezium Connector

通常，Oracle 服务器上的 redo 日志配置为不保留数据库的完整历史记录。因此，Debezium Oracle 连接器无法从日志检索数据库的完整历史记录。要让连接器为数据库的当前状态建立基准，连接器首次启动时，它会执行数据库的初始 一致快照。

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

当快照模式被设置为默认模式时，连接器会完成以下任务来创建快照：

生成的初始快照捕获捕获捕获的表中每行的当前状态。在这个基准状态中，连接器会捕获后续更改。

在快照进程开始后，如果进程因为连接器失败、重新平衡或其他原因而中断，则进程会在连接器重启后重启。连接器完成初始快照后，它会继续从在第 3 步中读取的位置进行流，使其不会错过任何更新。如果连接器因为任何原因而再次停止，它会在重启后从之前关闭的位置恢复流更改。

如需更多信息，请参阅连接器配置属性表中的 snapshot.mode。

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

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

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

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

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

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

如需更多信息，请参阅 增加快照。

如需更多信息，请参阅 阻止快照。

为了提供管理快照的灵活性，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 <data-collection> WHERE <filter> ....

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

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

INSERT INTO myschema.debezium_signal (id, type, data) VALUES('ad-hoc-1', 'execute-snapshot', '{"data-collections": ["schema1.products"],"type":"incremental", "additional-conditions":[{"data-collection": "schema1.products", "filter": "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-conditions": [{"data-collection": "schema1.products" ,"filter":"color='blue'"}]}}`

Key = `test_connector`

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

Key = `test_connector`

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

  {"type": "blocking", "data-collections": ["schema1.table1", "schema1.table2"], "additional-conditions": [{"data-collection": "schema1.table1", "filter": "SELECT * FROM [schema1].[table1] WHERE column1 = 0 ORDER BY column2 DESC"}, {"data-collection": "schema1.table2", "filter": "SELECT * FROM [schema1].[table2] WHERE column2 > 0"}]}

默认情况下，Oracle 连接器将所有 INSERT、UPDATE 和 DELETE 操作的更改事件写入特定于该表的单一 Apache Kafka 主题。连接器使用以下惯例来命名更改事件主题：

topicPrefix.schemaName.tableName

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

例如，如果 fulfillment 是服务器名称，inventory 是 schema 名称，数据库包括名为 orders, customers, 和 products 的表，Debezium Oracle 连接器会向以下 Kafka 主题发送事件，数据库中的每个表有一个。

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

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

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

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

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

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

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

您可以配置 Debezium Oracle 连接器来生成模式更改事件，该事件描述了应用到数据库中表的结构更改。连接器将模式更改事件写入名为 < serverName&gt; 的 Kafka 主题，其中 topicName 是 topic.prefix 配置属性中指定的命名空间。

当 Debezium 从新表中流数据或更改表结构时，Debezium 会向 schema 更改主题发送一个新消息。

连接器发送到 schema 更改主题的消息包含一个有效负载，以及可选的包含更改事件消息的 schema。

模式更改事件的 schema 具有以下元素：

{
  "schema": {
    "type": "struct",
    "fields": [
      {
        "type": "string",
        "optional": false,
        "field": "databaseName"
      }
    ],
    "optional": false,
    "name": "io.debezium.connector.oracle.SchemaChangeKey",
    "version": 1
  },
  "payload": {
    "databaseName": "inventory"
  }
}

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

{
  "schema": {
  ...
  },
  "payload": {
    "source": {
      "version": "2.5.4.Final",
      "connector": "oracle",
      "name": "server1",
      "ts_ms": 1588252618953,
      "snapshot": "true",
      "db": "ORCLPDB1",
      "schema": "DEBEZIUM",
      "table": "CUSTOMERS",
      "txId" : null,
      "scn" : "1513734",
      "commit_scn": "1513754",
      "lcr_position" : null,
      "rs_id": "001234.00012345.0124",
      "ssn": 1,
      "redo_thread": 1,
      "user_name": "user"
    },
    "ts_ms": 1588252618953, 1
    "databaseName": "ORCLPDB1", 2
    "schemaName": "DEBEZIUM", //
    "ddl": "CREATE TABLE \"DEBEZIUM\".\"CUSTOMERS\" \n   (    \"ID\" NUMBER(9,0) NOT NULL ENABLE, \n    \"FIRST_NAME\" VARCHAR2(255), \n    \"LAST_NAME" VARCHAR2(255), \n    \"EMAIL\" VARCHAR2(255), \n     PRIMARY KEY (\"ID\") ENABLE, \n     SUPPLEMENTAL LOG DATA (ALL) COLUMNS\n   ) SEGMENT CREATION IMMEDIATE \n  PCTFREE 10 PCTUSED 40 INITRANS 1 MAXTRANS 255 \n NOCOMPRESS LOGGING\n  STORAGE(INITIAL 65536 NEXT 1048576 MINEXTENTS 1 MAXEXTENTS 2147483645\n  PCTINCREASE 0 FREELISTS 1 FREELIST GROUPS 1\n  BUFFER_POOL DEFAULT FLASH_CACHE DEFAULT CELL_FLASH_CACHE DEFAULT)\n  TABLESPACE \"USERS\" ", 3
    "tableChanges": [ 4
      {
        "type": "CREATE", 5
        "id": "\"ORCLPDB1\".\"DEBEZIUM\".\"CUSTOMERS\"", 6
        "table": { 7
          "defaultCharsetName": null,
          "primaryKeyColumnNames": [ 8
            "ID"
          ],
          "columns": [ 9
            {
              "name": "ID",
              "jdbcType": 2,
              "nativeType": null,
              "typeName": "NUMBER",
              "typeExpression": "NUMBER",
              "charsetName": null,
              "length": 9,
              "scale": 0,
              "position": 1,
              "optional": false,
              "autoIncremented": false,
              "generated": false
            },
            {
              "name": "FIRST_NAME",
              "jdbcType": 12,
              "nativeType": null,
              "typeName": "VARCHAR2",
              "typeExpression": "VARCHAR2",
              "charsetName": null,
              "length": 255,
              "scale": null,
              "position": 2,
              "optional": false,
              "autoIncremented": false,
              "generated": false
            },
            {
              "name": "LAST_NAME",
              "jdbcType": 12,
              "nativeType": null,
              "typeName": "VARCHAR2",
              "typeExpression": "VARCHAR2",
              "charsetName": null,
              "length": 255,
              "scale": null,
              "position": 3,
              "optional": false,
              "autoIncremented": false,
              "generated": false
            },
            {
              "name": "EMAIL",
              "jdbcType": 12,
              "nativeType": null,
              "typeName": "VARCHAR2",
              "typeExpression": "VARCHAR2",
              "charsetName": null,
              "length": 255,
              "scale": null,
              "position": 4,
              "optional": false,
              "autoIncremented": false,
              "generated": false
            }
          ],
          "attributes": [ 10
            {
              "customAttribute": "attributeValue"
            }
          ]
        }
      }
    ]
  }
}

在连接器发送到 schema 更改主题的消息中，message 键是包含 schema 更改的数据库的名称。在以下示例中，payload 字段包含 databaseName 键：

{
  "schema": {
    "type": "struct",
    "fields": [
      {
        "type": "string",
        "optional": false,
        "field": "databaseName"
      }
    ],
    "optional": false,
    "name": "io.debezium.connector.oracle.SchemaChangeKey",
    "version": 1
  },
  "payload": {
    "databaseName": "ORCLPDB1"
  }
}

Debezium 可以生成代表事务元数据边界的事件，并增强数据更改事件消息。

数据库事务由语句块表示，该块包含在 BEGIN 和 END 关键字之间。Debezium 为每个事务中的 BEGIN 和 END 分隔符生成事务边界事件。事务边界事件包含以下字段：

以下示例显示了典型的事务边界消息：

{
  "status": "BEGIN",
  "id": "5.6.641",
  "ts_ms": 1486500577125,
  "event_count": null,
  "data_collections": null
}

{
  "status": "END",
  "id": "5.6.641",
  "ts_ms": 1486500577691,
  "event_count": 2,
  "data_collections": [
    {
      "data_collection": "ORCLPDB1.DEBEZIUM.CUSTOMER",
      "event_count": 1
    },
    {
      "data_collection": "ORCLPDB1.DEBEZIUM.ORDER",
      "event_count": 1
    }
  ]
}

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

{
  "before": null,
  "after": {
    "pk": "2",
    "aa": "1"
  },
  "source": {
...
  },
  "op": "c",
  "ts_ms": "1580390884335",
  "transaction": {
    "id": "5.6.641",
    "total_order": "1",
    "data_collection_order": "1"
  }
}

Oracle 按照它们发生的顺序将所有更改写入 redo 日志，包括回滚后丢弃的更改。因此，来自独立事务的并发更改会被干预。当连接器首次读取更改流时，因为它无法立即决定提交或回滚哪些更改，它会临时将更改事件存储在内部缓冲区中。提交更改后，连接器会将更改事件从缓冲区写入 Kafka。连接器丢弃回滚丢弃的更改事件。

您可以通过设置属性 log.mining.buffer.type 来配置连接器使用的缓冲机制。

当 Debezium Oracle 连接器被配置为使用 LogMiner 时，它会使用基于系统更改号(SCN)的开始和结束范围从 Oracle 收集更改事件。连接器会自动管理这个范围，根据连接器是否能够流向实时流更改，或者因为数据库中大或批量事务的卷而处理更改，以自动增加或减少范围。

在某些情况下，Oracle 数据库会以一个非常高的数量来提升 SCN，而不是以恒定率增加 SCN 值。由于特定集成与数据库交互方式或出现热备份等事件，所以可能会出现 SCN 值的跳过。

Debezium Oracle 连接器依赖于以下配置属性来检测 SCN 差距并调整 mining 范围。

连接器首先比较当前 SCN 和当前 mining 范围内的最高 SCN 之间的变化数量。如果当前 SCN 值和最高 SCN 值之间的区别大于最小空白大小，则连接器可能会检测到 SCN 差距。要确认是否存在差距，连接器会随后比较当前 SCN 和 SCN 在前一个 mining 范围末尾的时间戳。如果时间戳之间的区别小于最大时间间隔，则确认存在 SCN 差距。

当发生 SCN 差距时，Debezium 连接器会自动使用当前的 SCN 作为当前 mining 会话范围的端点。这允许连接器在没有返回任何更改之间快速捕获实时事件，因为 SCN 值意外增加。当连接器执行前面的步骤以响应 SCN 差距时，它会忽略 log.mining.batch.size.max 属性指定的值。在连接器完成 mining 会话并捕获到实时事件后，它会恢复最大日志最小批处理大小的强制。

Debezium Oracle 连接器跟踪连接器偏移中的系统更改号，以便在连接器重启时，它可以从离开的位置开始。这些偏移是每个发出的更改事件的一部分；但是，当数据库更改的频率较低（几小时或天）时，偏移可能会过时，并阻止连接器在事务日志中不再可用时成功重启。

对于使用非CDB 模式连接到 Oracle 的连接器，您可以启用 heartbeat.interval.ms 来强制连接器定期发出心跳事件，以便偏移保持同步。

对于使用 CDB 模式连接到 Oracle 的连接器，维护同步更为复杂。不仅必须设置 heartbeat.interval.ms，还需要设置 heartbeat.action.query。需要指定这两个属性，因为在 CDB 模式中，连接器仅跟踪 PDB 中的更改。需要补充机制来从可插拔数据库内触发更改事件。定期，心跳操作查询会导致连接器插入一个新的表行，或者在可插拔数据库中更新现有行。Debezium 检测到表更改，并为它们发出更改事件，确保偏移保持同步，即使在可插拔数据库中，进程不经常更改。