第 6 章 Oracle 的 Debezium Connector

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

您可以通过设置 snapshot.mode 连接器配置属性的值来自定义连接器创建快照的方法。默认情况下，连接器的快照模式被设置为 initial。

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

如需更多信息，请参阅连接器配置属性表中的 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
}

默认情况下，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 自定义主题命名的更多信息，请参阅 主题路由。

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

每当从新表流传输数据时，Debezium 会将新消息发送到此主题。

连接器发送到 schema 更改主题的消息包含一个有效负载，也可以包含更改事件消息的 schema。模式更改事件消息的有效负载包括以下元素：

{
  "schema": {
  ...
  },
  "payload": {
    "source": {
      "version": "2.1.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 键是包含架构更改的数据库的名称。在以下示例中，payload 字段包含 databaseName 键：

{
  "schema": {
    "type": "struct",
    "fields": [
      {
        "type": "string",
        "optional": false,
        "field": "databaseName"
      }
    ],
    "optional": false,
    "name": "io.debezium.connector.oracle.SchemaChangeKey"
  },
  "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 值。SCN 值中的这个跳过可能会因为特定集成与数据库交互的方式发生，或者作为热备份等事件的结果。

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

连接器首先比较当前 SCN 和当前最小范围内的最高 SCN 之间的更改数的不同。如果当前 SCN 值和最高 SCN 值之间的差别大于最小差距大小，那么连接器可能会检测到 SCN 差距。要确认是否存在差距，连接器 接下来比较当前 SCN 和之前最小范围末尾的 SCN 的时间戳。如果时间戳之间的区别小于最大时间间隔，则确认存在 SCN 差距。

当发生 SCN 差距时，Debezium 连接器会自动使用当前的 SCN 作为当前最小会话范围的端点。这允许连接器快速捕获实时事件，而不会在返回不小的范围的情况下快速捕获实时事件，因为 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 会检测到表更改，并为它们发出更改事件，确保偏移保持同步，即使在处理经常更改的可插拔数据库中也是如此。