CDC框架之Debezium

简介

Debezium 是一组分布式服务，用于捕获数据库中的更改，以便您的应用程序可以看到这些更改并对其做出响应。Debezium 在更改事件流中记录每个数据库表中的所有行级更改，应用程序只需读取这些流以查看更改事件发生的相同顺序。

Debezium 构建在Apache Kafka之上，并提供与Kafka Connect兼容的连接器，用于监控特定的数据库管理系统。Debezium 在 Kafka 日志中记录数据更改的历史记录，从您的应用程序使用它们的位置。这使您的应用程序可以轻松正确且完整地使用所有事件。即使您的应用程序意外停止，它也不会错过任何内容：当应用程序重新启动时，它将继续使用它停止的事件。

Kafka Connect

Kafka Connect是一种用于在Kafka和其他系统之间可扩展的、可靠的流式传输数据的工具。它使得能够快速定义将大量数据集合移入和移出Kafka的连接器变得简单。 Kafka Connect可以获取整个数据库或从所有应用程序服务器收集指标到Kafka主题，使数据可用于低延迟的流处理。导出作业可以将数据从Kafka topic传输到二次存储和查询系统，或者传递到批处理系统以进行离线分析。 Kafka Connect功能包括：

• Kafka connector通用框架,提供统一的集成API
• 同时支持分布式模式和单机模式
• REST 接口，用来查看和管理Kafka connectors
• 自动化的offset管理，开发人员不必担心错误处理的影响
• 分布式、可扩展
• 流/批处理集成

KafkaCnnect有两个核心概念：Source和Sink。 Source负责导入数据到Kafka，Sink负责从Kafka导出数据，它们都被称为Connector。

Kafka connect的几个重要的概念包括：connectors、tasks、workers和converters。

• Connectors-通过管理任务来细条数据流的高级抽象

• Tasks- 数据写入kafka和数据从kafka读出的实现

• Workers-运行connectors和tasks的进程

• Converters- kafka connect和其他存储系统直接发送或者接受数据之间转换数据

  1. Connectors：在kafka connect中，connector决定了数据应该从哪里复制过来以及数据应该写入到哪里去，一个connector实例是一个需要负责在kafka和其他系统之间复制数据的逻辑作业，connector plugin是jar文件，实现了kafka定义的一些接口来完成特定的任务。

  2. Tasks：task是kafka connect数据模型的主角，每一个connector都会协调一系列的task去执行任务，connector可以把一项工作分割成许多的task，然后再把task分发到各个worker中去执行（分布式模式下），task不自己保存自己的状态信息，而是交给特定的kafka 主题去保存（config.storage.topic 和status.storage.topic）。在分布式模式下有一个概念叫做任务再平衡（Task Rebalancing），当一个connector第一次提交到集群时，所有的worker都会做一个task rebalancing从而保证每一个worker都运行了差不多数量的工作，而不是所有的工作压力都集中在某个worker进程中，而当某个进程挂了之后也会执行task rebalance。

  3. Workers:connectors和tasks都是逻辑工作单位，必须安排在进程中执行，而在kafka connect中，这些进程就是workers，分别有两种worker：standalone和distributed。这里不对standalone进行介绍，具体的可以查看官方文档。我个人觉得distributed worker很棒，因为它提供了可扩展性以及自动容错的功能，你可以使用一个group.ip来启动很多worker进程，在有效的worker进程中它们会自动的去协调执行connector和task，如果你新加了一个worker或者挂了一个worker，其他的worker会检测到然后在重新分配connector和task。

  4. Converters： converter会把bytes数据转换成kafka connect内部的格式，也可以把kafka connect内部存储格式的数据转变成bytes，converter对connector来说是解耦的，所以其他的connector都可以重用，例如，使用了avro converter，那么jdbc connector可以写avro格式的数据到kafka，当然，hdfs connector也可以从kafka中读出avro格式的数据。

Kafka Connect启动方式

• standalone

  bin/connect-standalone.shconfig/connect-standalone.properties connector1.properties[connector2.properties ...]
  

• distributed(集群模式)

  bin/connect-distributed.shconfig/connect-distributed.properties
  

• 容器环境

启动示例

启动zookeeper

docker run -it --rm --name zookeeper -p 2181:2181 -p 2888:2888 -p 3888:3888 debezium/zookeeper:1.7

启动kafka

$ docker run -it --rm --name kafka -p 9092:9092 --link zookeeper:zookeeper debezium/kafka:1.7

启动一个被监控的数据库(MySQL)

$ docker run -it --rm --name mysql -p 3306:3306 -e MYSQL_ROOT_PASSWORD=debezium -e MYSQL_USER=mysqluser -e MYSQL_PASSWORD=mysqlpw debezium/example-mysql:1.7

启动Kafka Connect

$ docker run -it --rm --name connect -p 8083:8083 -e GROUP_ID=1 -e CONFIG_STORAGE_TOPIC=my_connect_configs -e OFFSET_STORAGE_TOPIC=my_connect_offsets -e STATUS_STORAGE_TOPIC=my_connect_statuses --link zookeeper:zookeeper --link kafka:kafka --link mysql:mysql debezium/connect:1.7

注册连接器

通过Debezium连接器部分中提供的参数，使用Kafka Connect提供的REST API注册连接器

• 使用curl:

  $ curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" localhost:8083/connectors/ -d '{ "name": "inventory-connector", "config": { "connector.class": "io.debezium.connector.mysql.MySqlConnector", "tasks.max": "1", "database.hostname": "mysql", "database.port": "3306", "database.user": "debezium", "database.password": "dbz", "database.server.id": "184054", "database.server.name": "dbserver1", "database.include.list": "inventory", "database.history.kafka.bootstrap.servers": "kafka:9092", "database.history.kafka.topic": "dbhistory.inventory" } }'
  

  Windows 用户可能需要转义双引号。例如：

  $ curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" localhost:8083/connectors/ -d '{ "name": "inventory-connector", "config": { "connector.class": "io.debezium.connector.mysql.MySqlConnector", "tasks.max": "1", "database.hostname": "mysql", "database.port": "3306", "database.user": "debezium", "database.password": "dbz", "database.server.id": "184054", "database.server.name": "dbserver1", "database.include.list": "inventory", "database.history.kafka.bootstrap.servers": "kafka:9092", "database.history.kafka.topic": "dbhistory.inventory" } }'
  

• example

  {
  "name": "inventory-connector",
  "config": {
  "connector.class": "io.debezium.connector.mysql.MySqlConnector",
  "database.hostname": "192.168.99.100",
  "database.port": "3306",
  "database.user": "debezium-user",
  "database.password": "debezium-user-pw",
  "database.server.id": "184054",
  "database.server.name": "fullfillment",
  "database.include.list": "inventory",
  "database.history.kafka.bootstrap.servers": "kafka:9092",
  "database.history.kafka.topic": "dbhistory.fullfillment",
  "include.schema.changes": "true"
  }
  }
  

验证连接器

curl -H "Accept:application/json" localhost:8083/connectors/

Debezium连接器

Debezium connector for MySQL

简介

​ MySQL 有一个二进制日志（binlog），它按照提交到数据库的顺序记录所有操作。这包括对表模式的更改以及对表中数据的更改。MySQL 使用 binlog 进行复制和恢复。

​ 该Debezium的MySQL连接器读取二进制日志，产生变化的事件行级INSERT，UPDATE和DELETE运营，并发出更改事件kafka的话题。客户端应用程序读取这些 Kafka 主题。

​ 由于 MySQL 通常设置为在指定的时间段后清除 binlog，因此 MySQL 连接器会为您的每个数据库执行初始一致的快照。MySQL 连接器从创建快照的点读取二进制日志。

官方文档:connector for MySQL

注册参数表

基础参数

高级参数

注册示例

{
"name": "inventory-connector",
"config": {
"connector.class": "io.debezium.connector.mysql.MySqlConnector",
"database.hostname": "192.168.99.100",
"database.port": "3306",
"database.user": "debezium-user",
"database.password": "debezium-user-pw",
"database.server.id": "184054",
"database.server.name": "fullfillment",
"database.include.list": "inventory",
"database.history.kafka.bootstrap.servers": "kafka:9092",
"database.history.kafka.topic": "dbhistory.fullfillment",
"include.schema.changes": "true"
}
}

事件Schame

可以使用参数converter.schemas.enable=false，设置数据不附带schame信息

INSERT

{
"schema": { ... },
"payload": {
"op": "c",
"ts_ms": 1465491411815,
"before": null,
"after": {
"id": 1004,
"first_name": "Anne",
"last_name": "Kretchmar",
"email": "annek@noanswer.org"
},
"source": {
"version": "1.7.1.Final",
"connector": "mysql",
"name": "mysql-server-1",
"ts_ms": 0,
"snapshot": false,
"db": "inventory",
"table": "customers",
"server_id": 0,
"gtid": null,
"file": "mysql-bin.000003",
"pos": 154,
"row": 0,
"thread": 7,
"query": "INSERT INTO customers (first_name, last_name, email) VALUES ('Anne', 'Kretchmar', 'annek@noanswer.org')"
}
}
}

UPDATE

{
"schema": { ... },
"payload": {
"before": {
"id": 1004,
"first_name": "Anne",
"last_name": "Kretchmar",
"email": "annek@noanswer.org"
},
"after": {
"id": 1004,
"first_name": "Anne Marie",
"last_name": "Kretchmar",
"email": "annek@noanswer.org"
},
"source": {
"version": "1.7.1.Final",
"name": "mysql-server-1",
"connector": "mysql",
"name": "mysql-server-1",
"ts_ms": 1465581029100,
"snapshot": false,
"db": "inventory",
"table": "customers",
"server_id": 223344,
"gtid": null,
"file": "mysql-bin.000003",
"pos": 484,
"row": 0,
"thread": 7,
"query": "UPDATE customers SET first_name='Anne Marie' WHERE id=1004"
},
"op": "u",
"ts_ms": 1465581029523
}
}

DELETE

{
"schema": { ... },
"payload": {
"before": {
"id": 1004,
"first_name": "Anne Marie",
"last_name": "Kretchmar",
"email": "annek@noanswer.org"
},
"after": null,
"source": {
"version": "1.7.1.Final",
"connector": "mysql",
"name": "mysql-server-1",
"ts_ms": 1465581902300,
"snapshot": false,
"db": "inventory",
"table": "customers",
"server_id": 223344,
"gtid": null,
"file": "mysql-bin.000003",
"pos": 805,
"row": 0,
"thread": 7,
"query": "DELETE FROM customers WHERE id=1004"
},
"op": "d",
"ts_ms": 1465581902461
}
}

SNAPSHOT

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

Debezium connector for MongoDB

简介

​ MongoDB 的复制机制提供了冗余和高可用性，是在生产中运行 MongoDB 的首选方式。MongoDB 连接器捕获副本集或分片集群中的更改。MongoDB副本集由一组服务器组成，这些服务器都具有相同数据的副本，并且复制确保客户端对副本集主服务器上的文档所做的所有更改都正确应用于其他副本集的服务器，称为辅助服务器。MongoDB 复制的工作原理是让主节点在其oplog（或操作日志）中记录更改，然后每个辅助节点读取主节点的 oplog 并按顺序将所有操作应用于它们自己的文档。将新服务器添加到副本集时，该服务器首先执行快照主数据库上的所有数据库和集合，然后读取主数据库的 oplog 以应用自它开始快照以来可能进行的所有更改。当这个新服务器赶上主服务器 oplog 的尾部时，它就变成了辅助服务器（并且能够处理查询）。

MongoDB 连接器使用相同的复制机制，尽管它实际上并未成为副本集的成员。然而，就像 MongoDB 辅助节点一样，连接器总是读取副本集主节点的 oplog。而且，当连接器第一次看到副本集时，它会查看 oplog 以获取最后记录的事务，然后执行主数据库和集合的快照。当所有数据都被复制后，连接器从它之前从 oplog 读取的位置开始流式传输更改。MongoDB oplog 中的操作是幂等的，因此无论应用多少次操作，它们都会导致相同的结束状态。

随着 MongoDB 连接器进程的变化，它会定期记录事件发生的 oplog 中的位置。当 MongoDB 连接器停止时，它会记录它处理的最后一个 oplog 位置，以便在重新启动时它只是从该位置开始流式传输。换句话说，连接器可以停止、升级或维护，并在一段时间后重新启动，它会从停止的地方恢复，而不会丢失任何事件。当然，MongoDB 的 oplog 通常有一个最大大小的上限，这意味着连接器不应停止太长时间，否则 oplog 中的某些操作可能会在连接器有机会读取它们之前被清除。在这种情况下，在重新启动时，连接器将检测丢失的 oplog 操作，执行快照，然后继续流式传输更改。

MongoDB 连接器也非常容忍副本集的成员资格和领导权的变化、分片集群中分片的添加或删除以及可能导致通信故障的网络问题。连接器始终使用副本集的主节点来流式传输更改，因此当副本集进行选举并且不同的节点成为主节点时，连接器将立即停止流式传输更改，连接到新的主节点，并使用新的主节点开始流式传输更改节点。同样，如果连接器在与主副本集通信时遇到任何问题，它将尝试重新连接（使用指数退避以避免网络或副本集不堪重负）并从上次停止的地方继续流式传输更改。

其他资源

• 复制机制

• 副本集

• 副本集选举

• 分片集群

• 分片添加

• 碎片移除

MongoDB 连接器的工作原理

连接器支持的 MongoDB 拓扑的概述对于规划您的应用程序很有用。

配置和部署 MongoDB 连接器时，它首先连接到种子地址处的 MongoDB 服务器，并确定有关每个可用副本集的详细信息。由于每个副本集都有自己独立的 oplog，因此连接器将尝试为每个副本集使用单独的任务。连接器可以限制它将使用的最大任务数，如果没有足够的任务可用，连接器将为每个任务分配多个副本集，尽管该任务仍将为每个副本集使用单独的线程。针对分片集群运行连接器时，请使用tasks.max大于副本集数量的值。这将允许连接器为每个副本集创建一个任务，并让 Kafka Connect 协调、分发和管理所有可用工作进程之间的任务。

注册参数表

基础参数

高级参数

注册示例

{
"name": "inventory-connector",
"config": {
"connector.class": "io.debezium.connector.mongodb.MongoDbConnector",
"mongodb.hosts": "rs0/192.168.99.100:27017",
"mongodb.name": "fullfillment",
"collection.include.list": "inventory[.]*",
}
}

事件schema

INSERT

{
"schema": { ... },
"payload": {
"after": "{"_id" : {"$numberLong" : "1004"},"first_name" : "Anne","last_name" : "Kretchmar","email" : "annek@noanswer.org"}",
"patch": null,
"source": {
"version": "1.7.1.Final",
"connector": "mongodb",
"name": "fulfillment",
"ts_ms": 1558965508000,
"snapshot": false,
"db": "inventory",
"rs": "rs0",
"collection": "customers",
"ord": 31,
"h": 1546547425148721999
},
"op": "c",
"ts_ms": 1558965515240
}
}

UPDATE

{
"schema": { ... },
"payload": {
"op": "u",
"ts_ms": 1465491461815,
"patch": "{"$set":{"first_name":"Anne Marie"}}",
"filter": "{"_id" : {"$numberLong" : "1004"}}",
"source": {
"version": "1.7.1.Final",
"connector": "mongodb",
"name": "fulfillment",
"ts_ms": 1558965508000,
"snapshot": true,
"db": "inventory",
"rs": "rs0",
"collection": "customers",
"ord": 6,
"h": 1546547425148721999
}
}
}

DELETE

{
"schema": { ... },
"payload": {
"op": "d",
"ts_ms": 1465495462115,
"filter": "{"_id" : {"$numberLong" : "1004"}}",
"source": {
"version": "1.7.1.Final",
"connector": "mongodb",
"name": "fulfillment",
"ts_ms": 1558965508000,
"snapshot": true,
"db": "inventory",
"rs": "rs0",
"collection": "customers",
"ord": 6,
"h": 1546547425148721999
}
}
}

Debezium connector for PostgreSQL

简介

Debezium 的 PostgreSQL 连接器捕获 PostgreSQL 数据库模式中的行级更改。支持 PostgreSQL 版本 9.6、10、11、12 和 13。

第一次连接到 PostgreSQL 服务器或集群时，连接器会拍摄所有模式的一致快照。快照完成后，连接器会持续捕获插入、更新和删除数据库内容以及提交到 PostgreSQL 数据库的行级更改。连接器生成数据更改事件记录并将它们流式传输到 Kafka 主题。对于每个表，默认行为是连接器将所有生成的事件流式传输到该表的单独 Kafka 主题。应用程序和服务使用来自该主题的数据更改事件记录。

PostgreSQL 的逻辑解码特性是在 9.4 版本中引入的。它是一种机制，允许提取提交到事务日志的更改，并在输出插件的帮助下以用户友好的方式处理这些更改。输出插件使客户端能够使用更改。

PostgreSQL 连接器包含两个主要部分，它们协同工作以读取和处理数据库更改：

• 一个逻辑解码输出插件。您可能需要安装您选择使用的输出插件。在运行 PostgreSQL 服务器之前，您必须配置一个使用您选择的输出插件的复制槽。插件可以是以下之一：
  • decoderbufs 基于 Protobuf 并由 Debezium 社区维护。
  • wal2json 基于 JSON 并由 wal2json 社区维护。
  • pgoutput是 PostgreSQL 10+ 中标准的逻辑解码输出插件。它由 PostgreSQL 社区维护，由 PostgreSQL 本身用于逻辑复制。此插件始终存在，因此无需安装其他库。Debezium 连接器将原始复制事件流直接解释为更改事件。
• 读取所选逻辑解码输出插件产生的更改的 Java 代码（实际的 Kafka Connect 连接器）。它使用 PostgreSQL 的流式复制协议，通过 PostgreSQL JDBC 驱动程序

连接器为捕获的每个行级插入、更新和删除操作生成一个更改事件，并为单独的 Kafka 主题中的每个表发送更改事件记录。客户端应用程序读取与感兴趣的数据库表对应的 Kafka 主题，并且可以对它们从这些主题接收到的每个行级事件做出反应。

PostgreSQL 通常会在一段时间后清除预写日志 (WAL) 段。这意味着连接器没有对数据库所做的所有更改的完整历史记录。因此，当 PostgreSQL 连接器首次连接到特定的 PostgreSQL 数据库时，它首先对每个数据库模式执行一致的快照。连接器完成快照后，它会从创建快照的确切点继续流式传输更改。这样，连接器从所有数据的一致视图开始，并且不会忽略在拍摄快照时所做的任何更改。

连接器可以容忍故障。当连接器读取更改并生成事件时，它会记录每个事件的 WAL 位置。如果连接器因任何原因（包括通信故障、网络问题或崩溃）而停止，则在重新启动时，连接器会继续读取上次停止的 WAL。这包括快照。如果连接器在快照期间停止，则连接器在重新启动时会开始一个新的快照。

连接器的工作原理

要优化配置和运行 Debezium PostgreSQL 连接器，了解连接器如何执行快照、流更改事件、确定 Kafka 主题名称以及使用元数据会很有帮助。

安全

要使用 Debezium 连接器从 PostgreSQL 数据库流式传输更改，连接器必须在数据库中以特定权限运行。尽管授予必要权限的一种方法是为用户提供superuser权限，但这样做可能会将您的 PostgreSQL 数据暴露给未经授权的访问。最好创建一个专用的 Debezium 复制用户，向其授予特定权限，而不是向 Debezium 用户授予过多的权限。

有关为 Debezium PostgreSQL 用户配置权限的更多信息，请参阅设置权限。有关 PostgreSQL 逻辑复制安全性的更多信息，请参阅PostgreSQL 文档。

快照

大多数 PostgreSQL 服务器配置为不在 WAL 段中保留数据库的完整历史记录。这意味着 PostgreSQL 连接器将无法通过仅读取 WAL 来查看数据库的整个历史记录。因此，连接器第一次启动时，它会执行数据库的初始一致快照。执行快照的默认行为包括以下步骤。您可以通过将snapshot.mode连接器配置属性设置为以外的值来更改此行为initial。

1. 使用SERIALIZABLE、READ ONLY、DEFERRABLE隔离级别启动事务，以确保此事务中的后续读取针对数据的单个一致版本。由于其他客户端的后续INSERT、UPDATE和DELETE操作而对数据进行的任何更改对于此事务都是不可见的。
2. 读取服务器事务日志中的当前位置。
3. 扫描数据库表和模式，READ为每一行生成一个事件并将该事件写入适当的表特定的 Kafka 主题。
4. 提交交易。
5. 在连接器偏移中记录快照的成功完成。

如果连接器出现故障、则连接器在重新启动时会开始一个新的快照。连接器完成其初始快照后，PostgreSQL 连接器从读取的位置继续流式传输。这确保连接器不会错过任何更新。如果连接器因任何原因再次停止，则在重新启动时，连接器会继续从先前停止的位置流式传输更改。

注册参数表

基础参数

高级参数

注册示例

{
"name": "fulfillment-connector",
"config": {
"connector.class": "io.debezium.connector.postgresql.PostgresConnector",
"database.hostname": "192.168.99.100",
"database.port": "5432",
"database.user": "postgres",
"database.password": "postgres",
"database.dbname" : "postgres",
"database.server.name": "fulfillment",
"table.include.list": "public.inventory"
}
}

事件Schema

INSERT

{
"schema": { ... },
"payload": {
"before": null,
"after": {
"id": 1,
"first_name": "Anne",
"last_name": "Kretchmar",
"email": "annek@noanswer.org"
},
"source": {
"version": "1.7.1.Final",
"connector": "postgresql",
"name": "PostgreSQL_server",
"ts_ms": 1559033904863,
"snapshot": true,
"db": "postgres",
"sequence": "["24023119","24023128"]",
"schema": "public",
"table": "customers",
"txId": 555,
"lsn": 24023128,
"xmin": null
},
"op": "c",
"ts_ms": 1559033904863
}
}

UPDATE

{
"schema": { ... },
"payload": {
"before": {
"id": 1
},
"after": {
"id": 1,
"first_name": "Anne Marie",
"last_name": "Kretchmar",
"email": "annek@noanswer.org"
},
"source": {
"version": "1.7.1.Final",
"connector": "postgresql",
"name": "PostgreSQL_server",
"ts_ms": 1559033904863,
"snapshot": false,
"db": "postgres",
"schema": "public",
"table": "customers",
"txId": 556,
"lsn": 24023128,
"xmin": null
},
"op": "u",
"ts_ms": 1465584025523
}
}

DELETE

{
"schema": { ... },
"payload": {
"before": {
"id": 1
},
"after": null,
"source": {
"version": "1.7.1.Final",
"connector": "postgresql",
"name": "PostgreSQL_server",
"ts_ms": 1559033904863,
"snapshot": false,
"db": "postgres",
"schema": "public",
"table": "customers",
"txId": 556,
"lsn": 46523128,
"xmin": null
},
"op": "d",
"ts_ms": 1465581902461
}
}

Debezium Connector for Oracle

简介

Debezium 的 Oracle 连接器捕获并记录 Oracle 服务器上数据库中发生的行级更改，包括在连接器运行时添加的表。您可以将连接器配置为针对架构和表的特定子集发出更改事件，或者忽略、屏蔽或截断特定列中的值。

Debezium 通过使用本机 LogMiner 数据库包或XStream API从 Oracle 摄取更改事件。虽然连接器可能适用于各种 Oracle 版本和版本，但只有 Oracle EE 12 和 19 已经过测试。

Oracle 连接器的工作原理

要优化配置和运行 Debezium Oracle 连接器，了解连接器如何执行快照、流更改事件、确定 Kafka 主题名称以及使用元数据会很有帮助。

快照

通常，Oracle 服务器上的重做日志配置为不保留数据库的完整历史记录。因此，Debezium Oracle 连接器无法从日志中检索数据库的整个历史记录。为了使连接器能够为数据库的当前状态建立基线，连接器第一次启动时，它会执行数据库的初始一致快照。

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

用于创建初始快照的默认连接器工作流程

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

1. 确定要捕获的表
2. 获取ROW SHARE MODE对每个受监控表的锁定，以防止在创建快照期间发生结构更改。Debezium 持有锁的时间很短。
3. 从服务器的重做日志中读取当前系统更改号 (SCN) 位置。
4. 捕获所有相关表的结构。
5. 释放在步骤 2 中获得的锁。
6. 在步骤 3 ( SELECT * FROM … AS OF SCN 123) 中读取的 SCN 位置扫描所有相关的数据库表和模式为有效，READ为每一行生成一个事件，然后将事件记录写入特定于表的 Kafka 主题。
7. 在连接器偏移中记录快照的成功完成。

快照进程开始后，如果由于connector失败、rebalancing等原因导致进程中断，则在connector重启后进程会重新启动。连接器完成初始快照后，它会从它在步骤 3 中读取的位置继续流式传输，以便不会错过任何更新。如果连接器因任何原因再次停止，则在重新启动后，它会从之前停止的位置恢复流式更改。

注册参数表

基础参数

高级参数

注册示例

{
"name": "inventory-connector",
"config": {
"connector.class" : "io.debezium.connector.oracle.OracleConnector",
"tasks.max" : "1",
"database.server.name" : "server1",
"database.user" : "c##dbzuser",
"database.password" : "dbz",
"database.url": "jdbc:oracle:thin:@(DESCRIPTION=(ADDRESS_LIST=(LOAD_BALANCE=OFF)(FAILOVER=ON)(ADDRESS=(PROTOCOL=TCP)(HOST=<oracle ip 1>)(PORT=1521))(ADDRESS=(PROTOCOL=TCP)(HOST=<oracle ip 2>)(PORT=1521)))(CONNECT_DATA=SERVICE_NAME=)(SERVER=DEDICATED)))",
"database.dbname" : "ORCLCDB",
"database.pdb.name" : "ORCLPDB1",
"database.history.kafka.bootstrap.servers" : "kafka:9092",
"database.history.kafka.topic": "schema-changes.inventory"
}
}

事件schema

INSERT

{
"schema": { ...
},
"payload": {
"before": null,
"after": {
"ID": 1004,
"FIRST_NAME": "Anne",
"LAST_NAME": "Kretchmar",
"EMAIL": "annek@noanswer.org"
},
"source": {
"version": "1.7.1.Final",
"name": "server1",
"ts_ms": 1520085154000,
"txId": "6.28.807",
"scn": "2122185",
"commit_scn": "2122185",
"snapshot": false
},
"op": "c",
"ts_ms": 1532592105975
}
}

UPDATE

{
"schema": { ... },
"payload": {
"before": {
"ID": 1004,
"FIRST_NAME": "Anne",
"LAST_NAME": "Kretchmar",
"EMAIL": "annek@noanswer.org"
},
"after": {
"ID": 1004,
"FIRST_NAME": "Anne",
"LAST_NAME": "Kretchmar",
"EMAIL": "anne@example.com"
},
"source": {
"version": "1.7.1.Final",
"name": "server1",
"ts_ms": 1520085811000,
"txId": "6.9.809",
"scn": "2125544",
"commit_scn": "2125544",
"snapshot": false
},
"op": "u",
"ts_ms": 1532592713485
}
}

DELETE

{
"schema": { ... },
"payload": {
"before": {
"ID": 1004,
"FIRST_NAME": "Anne",
"LAST_NAME": "Kretchmar",
"EMAIL": "anne@example.com"
},
"after": null,
"source": {
"version": "1.7.1.Final",
"name": "server1",
"ts_ms": 1520085153000,
"txId": "6.28.807",
"scn": "2122184",
"commit_scn": "2122184",
"snapshot": false
},
"op": "d",
"ts_ms": 1532592105960
}
}

Debezium connector for SQL Server

简介

Debezium SQL Server 连接器捕获 SQL Server 数据库架构中发生的行级更改。

Debezium SQL Server 连接器第一次连接到 SQL Server 数据库或群集时，它会拍摄数据库中架构的一致快照。初始快照完成后，连接器连续捕获行级的变化INSERT，UPDATE或者DELETE是承诺到为CDC启用了SQL Server数据库操作。连接器为每个数据更改操作生成事件，并将它们流式传输到 Kafka 主题。连接器将表的所有事件流式传输到专用的 Kafka 主题。然后，应用程序和服务可以使用来自该主题的数据更改事件记录。

Debezium SQL Server 连接器基于SQL Server 2016 Service Pack 1 (SP1) 和更高版本的标准版或企业版中 提供的更改数据捕获功能。SQL Server 捕获进程监视指定的数据库和表，并将更改存储到专门创建的具有存储过程外观的更改表中。

要启用 Debezium SQL Server 连接器来捕获数据库操作的更改事件记录，您必须首先在 SQL Server 数据库上启用更改数据捕获。必须在数据库和要捕获的每个表上启用 CDC。当您设置CDC在源数据库上，该连接器可以捕捉行级INSERT，UPDATE以及DELETE发生在数据库操作。连接器将每个源表的事件记录写入一个专用于该表的 Kafka 主题。每个捕获的表都存在一个主题。客户端应用程序读取它们遵循的数据库表的 Kafka 主题，并可以响应它们从这些主题中消费的行级事件。

连接器第一次连接到 SQL Server 数据库或集群时，它会为所有配置为捕获更改的表获取架构的一致快照，并将此状态流式传输到 Kafka。快照完成后，连接器会持续捕获后续发生的行级更改。通过首先建立所有数据的一致视图，连接器可以继续读取，而不会丢失在快照发生时所做的任何更改。

Debezium SQL Server 连接器可以容忍失败。当连接器读取更改并生成事件时，它会定期在数据库日志中记录事件的位置（LSN / 日志序列号）。如果连接器因任何原因（包括通信故障、网络问题或崩溃）而停止，则在重新启动后，连接器将继续从它读取的最后一点开始读取 SQL Server CDC表。

SQL Server 连接器的工作原理

要优化配置和运行 Debezium SQL Server 连接器，了解连接器如何执行快照、流更改事件、确定 Kafka 主题名称以及使用元数据会很有帮助。

快照

QL Server CDC 并非旨在存储数据库更改的完整历史记录。为了让 Debezium SQL Server 连接器为数据库的当前状态建立基线，它使用了一个称为快照的过程。

您可以配置连接器创建快照的方式。默认情况下，连接器的快照模式设置为initial。基于这种initial快照模式，连接器第一次启动时，它会执行数据库的初始一致性快照。此初始快照捕获与为连接器配置的include和exclude属性（例如，table.include.list、column.include.list、table.exclude.list等）定义的条件匹配的任何表的结构和数据。

当连接器创建快照时，它完成以下任务：

1. 确定要捕获的表。
2. 获取对启用了 CDC 的 SQL Server 表的锁定，以防止在创建快照期间发生结构更改。锁的级别由snapshot.isolation.mode配置选项决定。
3. 读取服务器事务日志中的最大日志序列号 (LSN) 位置。
4. 捕获所有相关表的结构。
5. 如有必要，释放在步骤 2 中获得的锁。在大多数情况下，锁只持有很短的时间。
6. 根据步骤3中读取的LSN位置扫描要捕获的SQL Server源表和模式，为表READ中的每一行生成一个事件，并将事件写入该表的Kafka主题。
7. 在连接器偏移中记录快照的成功完成。

生成的初始快照捕获为 CDC 启用的表中每一行的当前状态。从此基线状态开始，连接器会在发生后续更改时捕获这些更改。

注册参数表

基础参数

高就参数

注册示例

{
"name": "inventory-connector",
"config": {
"connector.class": "io.debezium.connector.sqlserver.SqlServerConnector",
"database.hostname": "192.168.99.100",
"database.port": "1433",
"database.user": "sa",
"database.password": "Password!",
"database.dbname": "testDB",
"database.server.name": "fullfillment",
"table.include.list": "dbo.customers",
"database.history.kafka.bootstrap.servers": "kafka:9092",
"database.history.kafka.topic": "dbhistory.fullfillment"
}
}

事件Schema

INSERT

{
"schema": { ...},
"payload": {
"before": null,
"after": {
"id": 1005,
"first_name": "john",
"last_name": "doe",
"email": "john.doe@example.org"
},
"source": {
"version": "1.7.1.Final",
"connector": "sqlserver",
"name": "server1",
"ts_ms": 1559729468470,
"snapshot": false,
"db": "testDB",
"schema": "dbo",
"table": "customers",
"change_lsn": "00000027:00000758:0003",
"commit_lsn": "00000027:00000758:0005",
"event_serial_no": "1"
},
"op": "c",
"ts_ms": 1559729471739
}
}

UPDATE

{
"schema": { ... },
"payload": {
"before": {
"id": 1005,
"first_name": "john",
"last_name": "doe",
"email": "john.doe@example.org"
},
"after": {
"id": 1005,
"first_name": "john",
"last_name": "doe",
"email": "noreply@example.org"
},
"source": {
"version": "1.7.1.Final",
"connector": "sqlserver",
"name": "server1",
"ts_ms": 1559729995937,
"snapshot": false,
"db": "testDB",
"schema": "dbo",
"table": "customers",
"change_lsn": "00000027:00000ac0:0002",
"commit_lsn": "00000027:00000ac0:0007",
"event_serial_no": "2"
},
"op": "u",
"ts_ms": 1559729998706
}
}

DELETE

{
"schema": { ... },
},
"payload": {
"before": { <>
"id": 1005,
"first_name": "john",
"last_name": "doe",
"email": "noreply@example.org"
},
"after": null,
"source": {
"version": "1.7.1.Final",
"connector": "sqlserver",
"name": "server1",
"ts_ms": 1559730445243,
"snapshot": false,
"db": "testDB",
"schema": "dbo",
"table": "customers",
"change_lsn": "00000027:00000db0:0005",
"commit_lsn": "00000027:00000db0:0007",
"event_serial_no": "1"
},
"op": "d",
"ts_ms": 1559730450205
}
}

Debezium connector for Db2

参考官网

Debezium Connector for Cassandra

参考官网

Debezium connector for Vitess

参考官网

Debezium 内置引擎

Debezium 连接器通常通过将它们部署到 Kafka Connect 服务并配置一个或多个连接器来监控上游数据库并为它们在上游数据库中看到的所有更改生成数据更改事件来操作。这些数据更改事件被写入 Kafka，在那里它们可以被许多不同的应用程序独立使用。Kafka Connect 提供出色的容错性和可扩展性，因为它作为分布式服务运行，并确保所有已注册和配置的连接器始终运行。例如，即使集群中的一个 Kafka Connect 端点出现故障，其余的 Kafka Connect 端点也会重新启动之前在现在终止的端点上运行的所有连接器，从而最大限度地减少停机时间并消除管理活动。

并非每个应用程序都需要这种级别的容错和可靠性，并且他们可能不想依赖 Kafka 代理和 Kafka Connect 服务的外部集群。相反，某些应用程序更愿意将Debezium 连接器直接嵌入到应用程序空间中。他们仍然想要相同的数据更改事件，但更喜欢让连接器将它们直接发送到应用程序，而不是将它们保存在 Kafka 中。

该debezium-api模块定义了一个小型 API，允许应用程序使用 Debezium 引擎轻松配置和运行 Debezium 连接器。

内置引擎文档

依赖关系

要使用 Debezium Engine 模块，请将模块添加debezium-api到应用程序的依赖项中。这个 API 在debezium-embedded模块中有一个开箱即用的实现，也应该添加到依赖项中。对于 Maven，这需要将以下内容添加到应用程序的 POM 中：

<dependency>
<groupId>io.debezium</groupId>
<artifactId>debezium-api</artifactId>
<version>${version.debezium}</version>
</dependency>
<dependency>
<groupId>io.debezium</groupId>
<artifactId>debezium-embedded</artifactId>
<version>${version.debezium}</version>
</dependency>

其中，${version.debezium}要么是Debezium的您正在使用的或更新的Maven的属性值中包含了Debezium版本字符串。

同样，为您的应用程序将使用的每个 Debezium 连接器添加依赖项。例如，可以将以下内容添加到您的应用程序的 Maven POM 文件中，以便您的应用程序可以使用 MySQL 连接器：

<dependency>
<groupId>io.debezium</groupId>
<artifactId>debezium-connector-mysql</artifactId>
<version>${version.debezium}</version>
</dependency>

或者对于 MongoDB 连接器：

<dependency>
<groupId>io.debezium</groupId>
<artifactId>debezium-connector-mongodb</artifactId>
<version>${version.debezium}</version>
</dependency>

本文档的其余部分描述了在您的应用程序中嵌入 MySQL 连接器。除了特定于连接器的配置、主题和事件之外，其他连接器的使用方式类似。

代码使用

您的应用程序需要为您要运行的每个连接器实例设置一个嵌入式引擎。该io.debezium.engine.DebeziumEngine<R>班作为各地的任何Debezium连接器一个易于使用的包装和完全管理连接器的生命周期。您DebeziumEngine使用其构建器 API创建实例，提供以下内容：

• 您希望接收消息的格式，例如 JSON、Avro 或 Kafka Connect SourceRecord （请参阅输出消息格式）
• 定义引擎和连接器环境的配置属性（可能从属性文件加载）
• 将为连接器产生的每个数据更改事件调用的方法

这是配置和运行嵌入式MySQL 连接器的代码示例：

// Define the configuration for the Debezium Engine with MySQL connector...
final Properties props = config.asProperties();
props.setProperty("name", "engine");
props.setProperty("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore");
props.setProperty("offset.storage.file.filename", "/tmp/offsets.dat");
props.setProperty("offset.flush.interval.ms", "60000");
/* begin connector properties */
props.setProperty("database.hostname", "localhost");
props.setProperty("database.port", "3306");
props.setProperty("database.user", "mysqluser");
props.setProperty("database.password", "mysqlpw");
props.setProperty("database.server.id", "85744");
props.setProperty("database.server.name", "my-app-connector");
props.setProperty("database.history",
"io.debezium.relational.history.FileDatabaseHistory");
props.setProperty("database.history.file.filename",
"/path/to/storage/dbhistory.dat");
// Create the engine with this configuration ...
try (DebeziumEngine<ChangeEvent<String, String>> engine = DebeziumEngine.create(Json.class)
.using(props)
.notifying(record -> {
System.out.println(record);
}).build()
) {
// Run the engine asynchronously ...
ExecutorService executor = Executors.newSingleThreadExecutor();
executor.execute(engine);
// Do something else or wait for a signal or an event
}
// Engine is stopped when the main code is finished

让我们更详细地研究这段代码，从我们在这里重复的前几行开始：

// Define the configuration for the Debezium Engine with MySQL connector...
final Properties props = config.asProperties();
props.setProperty("name", "engine");
props.setProperty("connector.class", "io.debezium.connector.mysql.MySqlConnector");
props.setProperty("offset.storage", "org.apache.kafka.connect.storage.FileOffsetBackingStore");
props.setProperty("offset.storage.file.filename", "/tmp/offsets.dat");
props.setProperty("offset.flush.interval.ms", 60000);

这将创建一个新的标准Properties对象来设置引擎所需的几个字段，而不管使用的是哪个连接器。第一个是引擎的名称，它将在由连接器及其内部状态生成的源记录中使用，因此请在您的应用程序中使用有意义的名称。该connector.class字段定义了扩展Kafka Connectorg.apache.kafka.connect.source.SourceConnector抽象类的类的名称；在这个例子中，我们指定 Debezium 的MySqlConnector类。

当 Kafka Connect 连接器运行时，它从源读取信息并定期记录“偏移量”，这些“偏移量”定义了它已处理的信息量。如果连接器重新启动，它将使用最后记录的偏移量来知道它应该恢复读取源信息中的哪个位置。由于连接器不知道或不关心如何存储偏移量，引擎提供一种方法来存储和恢复这些偏移量。我们配置的接下来的几个字段指定我们的引擎应该使用FileOffsetBackingStore该类将偏移量存储在/path/to/storage/offset.dat本地文件系统上的文件（该文件可以任意命名并存储在任何地方）。此外，虽然连接器记录了它产生的每个源记录的偏移量，但引擎会定期将偏移量刷新到后备存储（在我们的例子中，每分钟一次）。这些字段可以根据您的应用程序的需要进行定制。

接下来的几行定义了特定于连接器的字段（记录在每个连接器的文档中），在我们的示例中是MySqlConnector连接器：

/* begin connector properties */
props.setProperty("database.hostname", "localhost")
props.setProperty("database.port", "3306")
props.setProperty("database.user", "mysqluser")
props.setProperty("database.password", "mysqlpw")
props.setProperty("database.server.id", "85744")
props.setProperty("database.server.name", "my-app-connector")
props.setProperty("database.history",
"io.debezium.relational.history.FileDatabaseHistory")
props.setProperty("database.history.file.filename",
"/path/to/storage/dbhistory.dat")

在这里，我们设置运行 MySQL 数据库服务器的主机名和端口号，并定义将用于连接到 MySQL 数据库的用户名和密码。请注意，对于 MySQL，用户名和密码应对应于已被授予以下 MySQL 权限的 MySQL 数据库用户：

• SELECT
• RELOAD
• SHOW DATABASES
• REPLICATION SLAVE
• REPLICATION CLIENT

读取数据库的一致快照时需要前三个权限。最后两个权限允许数据库读取服务器的binlog，通常用于MySQL复制。

该配置还包括一个数字标识符server.id。由于 MySQL 的 binlog 是 MySQL 复制机制的一部分，为了读取 binlog，MySqlConnector实例必须加入 MySQL 服务器组，这意味着该服务器 ID在组成 MySQL 服务器组的所有进程中必须是唯一的，并且是介于两者之间的任意整数1 和 2 32 -1。在我们的代码中，我们将它设置为一个相当大但有些随机的值，我们将仅用于我们的应用程序。

该配置还为 MySQL 服务器指定了一个逻辑名称。连接器在它生成的每个源记录的主题字段中包含此逻辑名称，使您的应用程序能够识别这些记录的来源。我们的示例使用服务器名称“products”，大概是因为数据库包含产品信息。当然，您可以将其命名为对您的应用程序有意义的任何名称。

当MySqlConnector该类运行时，它会读取 MySQL 服务器的二进制日志，其中包括对服务器托管的数据库所做的所有数据更改和架构更改。由于对数据的所有更改都是根据记录更改时所属表的架构来构建的，因此连接器需要跟踪所有架构更改，以便它可以正确解码更改事件。连接器记录模式信息，以便连接器重新启动并从上次记录的偏移量继续读取时，它确切地知道该偏移量处的数据库模式是什么样的。在我们配置的最后两个字段中定义了连接器如何记录数据库模式历史，即我们的连接器应该使用FileDatabaseHistory该类来存储数据库模式历史更改/path/to/storage/dbhistory.dat 本地文件系统上的文件（同样，此文件可以命名为任何名称并存储在任何位置）。

最后使用该build()方法构建不可变配置。（顺便说一句，我们可以使用其中一种方法从属性文件中读取配置，而不是以编程方式构建它Configuration.read(…)。）

现在我们有了一个配置，我们可以创建我们的引擎。这里再次是相关的代码行：

// Create the engine with this configuration ...
try (DebeziumEngine<ChangeEvent<String, String>> engine = DebeziumEngine.create(Json.class)
.using(props)
.notifying(record -> {
System.out.println(record);
})
.build()) {
}

所有更改事件都将传递给给定的处理程序方法，该方法必须匹配java.util.function.Consumer<R>功能接口的签名，其中<R>必须匹配调用时指定的格式类型create()。请注意，您的应用程序的处理程序函数不应抛出任何异常；如果是，引擎将记录该方法抛出的任何异常，并将继续对下一个源记录进行操作，但您的应用程序将没有机会处理导致异常的特定源记录，这意味着您的应用程序可能会变得不一致与数据库。

此时，我们有一个已DebeziumEngine配置并准备运行的现有对象，但它不执行任何操作。将DebeziumEngine被设计成由一个异步执行Executor或ExecutorService：

// Run the engine asynchronously ...
ExecutorService executor = Executors.newSingleThreadExecutor();
executor.execute(engine);
// Do something else or wait for a signal or an event

您的应用程序可以通过调用其close()方法来安全优雅地停止引擎：

// At some later time ...
engine.close();

或者当引擎支持Closeable接口时，它会在try块离开时自动调用。

引擎的连接器将停止从源系统读取信息，将所有剩余的更改事件转发到您的处理程序函数，并将最新的偏移量刷新到偏移存储。只有在所有这些完成之后，引擎的run()方法才会返回。如果您的应用程序需要在退出之前等待引擎完全停止，您可以使用ExcecutorService shutdown和awaitTermination方法执行此操作：

try {
executor.shutdown();
while (!executor.awaitTermination(5, TimeUnit.SECONDS)) {
logger.info("Waiting another 5 seconds for the embedded engine to shut down");
}
}
catch ( InterruptedException e ) {
Thread.currentThread().interrupt();
}

或者，您可以CompletionCallback在创建时注册DebeziumEngine为在引擎终止时通知的回调。

回想一下，当 JVM 关闭时，它只等待守护线程。因此，如果您的应用程序退出，请确保等待引擎完成或在守护线程上运行引擎。

您的应用程序应该始终正确地停止引擎以确保正常和完全关闭，并且每个源记录都准确地发送到应用程序一次。例如，不要依赖于关闭ExecutorService，因为这会中断正在运行的线程。虽然DebeziumEngine当它的线程被中断时确实会终止，但引擎可能不会干净地终止，并且当您的应用程序重新启动时，它可能会看到一些在关闭之前处理过的相同源记录。

输出消息格式

DebeziumEngine#create()可以接受多个不同的参数，这些参数会影响消费者接收消息的格式。允许的值为：

• Connect.class - 输出值是包装 Kafka Connect 的更改事件 SourceRecord
• Json.class- 输出值是一对编码为JSON字符串的键和值
• Avro.class - 输出值是一对键值对，编码为 Avro 序列化记录
• CloudEvents.class- 输出值是一对键值对，编码为Cloud Events消息

在内部，引擎使用适当的 Kafka Connect 转换器实现，转换委托给该转换器实现。可以使用引擎属性对转换器进行参数化以修改其行为。

JSON输出格式的一个例子是

final Properties props = new Properties();
...
props.setProperty("converter.schemas.enable", "false"); // don't include schema in message
...
final DebeziumEngine<ChangeEvent<String, String>> engine = DebeziumEngine.create(Json.class)
.using(props)
.notifying((records, committer) -> {
for (ChangeEvent<String, String> r : records) {
System.out.println("Key = '" + r.key() + "' value = '" + r.value() + "'");
committer.markProcessed(r);
}
...

其中ChangeEvent数据类型是键/值对。

消息转换

在将消息传递给处理程序之前，可以通过 Kafka Connect简单消息转换(SMT)管道运行它们 。每个 SMT 都可以不变地传递消息、修改它或过滤掉它。该链是使用 property 配置的transforms。该属性包含要应用的转换的逻辑名称的逗号分隔列表。transforms.<logical_name>.type然后，属性transforms.<logical_name>.*为传递给转换的每个转换和配置选项定义实现类的名称。

配置的一个例子是

final Properties props = new Properties();
...
props.setProperty("transforms", "filter, router");
// (1)
props.setProperty("transforms.router.type", "org.apache.kafka.connect.transforms.RegexRouter");
// (2)
props.setProperty("transforms.router.regex", "(.*)");
// (3)
props.setProperty("transforms.router.replacement", "trf$1");
// (3)
props.setProperty("transforms.filter.type", "io.debezium.embedded.ExampleFilterTransform");
// (4)

1. 定义了两个转换 -filter和router
2. router转换的实现是org.apache.kafka.connect.transforms.RegexRouter
3. 该router改造有两种配置选择-regex和replacement
4. filter转换的实现是io.debezium.embedded.ExampleFilterTransform

高级记录消费

对于某些用例，例如尝试批量写入记录或针对异步 API 写入记录时，上述功能接口可能具有挑战性。在这些情况下，使用io.debezium.engine.DebeziumEngine.ChangeConsumer<R>.界面可能会更容易。

此接口具有单一功能，具有以下签名：

/**
* Handles a batch of records, calling the {@link RecordCommitter#markProcessed(Object)}
* for each record and {@link RecordCommitter#markBatchFinished()} when this batch is finished.
* @param records the records to be processed
* @param committer the committer that indicates to the system that we are finished
*/
void handleBatch(List<R> records, RecordCommitter<R> committer) throws InterruptedException;

正如 Javadoc 中所提到的，RecordCommitter每个记录都会调用该对象，并且一旦每个批次完成。该RecordCommitter接口是线程安全的，允许灵活处理记录。

您可以选择覆盖已处理记录的偏移量。这是通过首先Offsets调用构建一个新 对象来完成的RecordCommitter#buildOffsets()，使用 更新偏移量Offsets#set(String key, Object value)，然后RecordCommitter#markProcessed(SourceRecord record, Offsets sourceOffsets)使用更新的Offsets.

要使用ChangeConsumerAPI，您必须将接口的实现传递给notifyingAPI，如下所示：

class MyChangeConsumer implements DebeziumEngine.ChangeConsumer<RecordChangeEvent<SourceRecord>> {
public void handleBatch(List<RecordChangeEvent<SourceRecord>> records, RecordCommitter<RecordChangeEvent<SourceRecord>> committer) throws InterruptedException {
...
}
}
// Create the engine with this configuration ...
DebeziumEngine<RecordChangeEvent<SourceRecord>> engine = DebeziumEngine.create(ChangeEventFormat.of(Connect.class))
.using(props)
.notifying(new MyChangeConsumer())
.build();

如果使用 JSON 格式（等效格式也适用于其他格式），则代码如下所示：

class JsonChangeConsumer implements DebeziumEngine.ChangeConsumer<ChangeEvent<String, String>> {
public void handleBatch(List<ChangeEvent<String, String>> records,
RecordCommitter<ChangeEvent<String, String>> committer) throws InterruptedException {
...
}
}
// Create the engine with this configuration ...
DebeziumEngine<ChangeEvent<String, String>> engine = DebeziumEngine.create(Json.class)
.using(props)
.notifying(new MyChangeConsumer())
.build();

引擎属性

除非默认值可用，否则需要以下配置属性（为了文本格式，Java 类的包名称替换为<…>）。

数据库历史属性

一些连接器还需要配置数据库历史的附加属性集：

• MySQL
• SQL Server
• Oracle
• Db2

如果没有正确配置数据库历史，连接器将拒绝启动。默认配置期望 Kafka 集群可用。对于其他部署，可以使用基于文件的数据库历史存储实现。

处理故障

当引擎执行时，它的连接器会主动记录每个源记录中的源偏移量，并且引擎会定期将这些偏移量刷新到持久存储中。当应用程序和引擎正常关闭或崩溃时，当它们重新启动时，引擎及其连接器将从上次记录的偏移量开始重新读取源信息。

那么，当您的应用程序在嵌入式引擎运行时出现故障时会发生什么？最终效果是应用程序可能会在重新启动后收到一些在崩溃前已经处理过的源记录。多少取决于引擎将偏移量刷新到其存储（通过offset.flush.interval.ms属性）的频率以及特定连接器在一批中返回的源记录数。最好的情况是每次都刷新偏移量（例如，offset.flush.interval.ms设置为 0），但即使如此，嵌入式引擎仍然只会在从连接器接收到每批源记录后才刷新偏移量。

例如，MySQL 连接器使用max.batch.size来指定批处理中可以出现的最大源记录数。即使offset.flush.interval.ms设置为 0，当应用程序在崩溃后重新启动时，它可能会看到最多n 个重复项，其中n是批次的大小。如果该offset.flush.interval.ms属性设置得更高，则应用程序可能会看到最多n * m重复项，其中n是批次的最大大小，而m是在单个偏移刷新间隔期间可能累积的批次数。（显然，可以将嵌入式连接器配置为不使用批处理并始终刷新偏移量，从而导致应用程序永远不会收到任何重复的源记录。然而，这会显着增加开销并降低连接器的吞吐量。）

底线是当使用嵌入式连接器时，应用程序将在正常操作期间（包括正常关闭后重新启动）接收每个源记录一次，但确实需要容忍在崩溃或不正确关闭后立即重新启动后接收重复事件. 如果应用程序需要更严格的一次性行为，那么它们应该使用可以提供一次性保证的完整 Debezium 平台（即使在崩溃和重启之后）。

最后

以上就是如意凉面为你收集整理的CDC框架之Debezium使用CDC框架之Debezium的全部内容，希望文章能够帮你解决CDC框架之Debezium使用CDC框架之Debezium所遇到的程序开发问题。

如果觉得靠谱客网站的内容还不错，欢迎将靠谱客网站推荐给程序员好友。