使用Kafka Connect和CDC保持多数据库同步

使用JDBC将数据从数据库移动到Apache Kafka

Kafka Connect用户通常对如何通过无代码/低代码的方式，使用连接器在Kafka中进行数据的写入和读取非常熟悉，但是用户可能并不知道Kafka Connect在解决一些复杂问题方面的强大功能和灵活性。

JDBC源端连接器可能是将数据从数据库抽取到Kafka的最简单方法。通过将此连接器与Kafka Connect结合使用，可以通过定期SQL查询批量获取数据，应用任何必要的转换，然后将其写入Kafka 主题。

该方法的优势该方法的局限性

使用CDC将数据从数据库移动到Kafka

下面是另一个方法，DBA可能不希望Kafka Connect不断使用SQL从数据库中提取数据，而且很多时候该数据库已经被其他更重要的任务所占据，无法承受Kafka Connect不断在数据库上运行SQL查询。

这里介绍一下CDC（变更数据捕获），这时Kafka Connect从数据库WAL（预写日志）文件中读取数据。从广义上讲，关系数据库都是使用事务日志的（也称为二进制日志或重做日志，具体取决于数据库厂商），数据库中的每个事件都会写入该日志。更新一行，插入一行，删除一行，所有这些都会写入数据库的事务日志。CDC工具的工作原理就是利用事务日志以低延迟和低负载的方式提取数据库（或其中的模式/表）上发生的事件。

该方法的优势该方法的局限性

使用Kafka Connect保持两个数据库同步

下面将深入探讨一个经典的用例，一家公司试图通过不断为其产品添加功能来满足客户的需求。

线上交易已经可用。来自在线交易的客户数据存储在PostgreSQL数据库中，但通过电话进行的交易保存在不同的数据库中。

传统的方法可能是让公司运行夜间作业，以保持两个数据库的同步，这种方法有各种限制，主要是：

使用Kafka Connect可以避免此问题。下面是高层架构的效果图：

在上面的示例中，数据是使用Debezium CDC连接器从两个数据库（SQL Server和PostgreSQL）中提取的。请注意这些表具有相似（但不相同）的数据模型，当然还有每个数据库特有的不同数据类型。

现在，两个接收端连接器将从对应的主题拉取以从另一个数据库接收数据。

仔细观察事件的顺序mssql数据库同步，你会发现它会陷入无限循环。原因如下：

应用将数据写入PostgreSQL；用于PostgreSQL的CDC连接器将该数据复制到Kafka主题，而JDBC接收端连接器将该Kafka主题的数据写入SQLServer数据库；该插入操作现在在SQLServer数据库上创建一个CDC创建事件，并写入一个Kafka主题，而JDBC接收端连接器将Kafka主题上存在的数据写入原始PostgreSQL，从而导致无限循环。

使用SMT可以打破该循环，下面会介绍。

在CDC源端连接器中使用SMT进行过滤

对于任何使用过Kafka Replicator的人来说，这都是一个熟悉的问题。Replicator支持多中心，生产者可以写入两个Kafka集群（DC-1和DC-2），并且一个集群的数据会被复制到另一个集群作为备份。通过配置provenance.header.enable=true，Replicator在复制后会将出处信息放在消息头中，如果目标集群的集群ID与出处消息头中的源集群ID匹配，并且目标主题名与出处消息头中的源主题名匹配，则Replicator将不会将消息复制到目标集群，通过这个方式，即可解决前述的事件重复问题。

现在已经讲解了Replicator如何处理双活场景，之后可能希望执行类似于Replicator的过滤。这时可以在数据库表中创建一列以显示该记录的创建位置，并在不需要时使用它来过滤。

例如，可以添加一个默认值为PGSQL的SRC列，以表示每条记录的首次创建位置（一旦将此记录复制到不同的数据库，此列对于可追溯性也变得很重要）。

启用过滤操作后可以将当前数据库（PostgreSQL）中生成的数据复制到目标端（SqlServer），但一旦复制到目标，它就不会发送回起始数据库（PostgreSQL），从而打破无限循环，以下是事件的顺序：

应用写入PostgreSQL，将记录插入到表中，并将SRC列中的数据设置为PGSQL（表示其来源，如上图所示）；用于PostgreSQL的Debezium CDC源端连接器将事件复制到主题；对SQLServer的插入操作由SQLServer的CDC连接器捕获。但是由于存在SRC=PGSQL的排除过滤器，因此将忽略此记录。

下面是PostgreSQL的CDC源端连接器的代码段，它过滤掉其他数据库生成的任何内容。（SRC == “SQLSRV”）

"name": "postgres-cdc-azure-src",
"config": {
"connector.class": "io.debezium.connector.postgresql.PostgresConnector",
"database.hostname": "oncloud.postgres.database.azure.com",
"transforms": "unwrap, filterPostGres, ExtractField, TimestampConverter",
"transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
"transforms.filterPostGres.type": "io.confluent.connect.transforms.Filter$Value",
"transforms.filterPostGres.filter.condition": "$[?(@.SRC == \"SQLSRV\")]",  
"transforms.filterPostGres.filter.type": "exclude",
"transforms.filterPostGres.missing.or.null.behavior": "include",

单消息转换（SMT）是在消息流经Kafka Connect时应用于消息的转换函数。SMT在源端连接器生成入站消息之后，在将入站消息写入Kafka之前对其进行转换。SMT还会在将出站消息发送到接收端连接器之前对其进行转换。

这里使用SMT（单消息转换）实现了该过滤器操作，以下是所有可用SMT的列表，并附有详细说明。开发者还需要考虑应用写入SQLServer时的情况。相关的源端连接器和接收端连接器的源代码在这里。

使用CDC和JDBC接收端连接器使三个数据库保持同步

现在已经知道使用Kafka Connect可以保持两个数据库同步，然后还可以实现三向数据库同步。

该用户现在希望为客户提供一项新功能，假设此数据存储在MySQL数据库中，现在必须保持三个数据库同步。

要实现这一需求，将需要另外两个连接器和更多的过滤：

过滤源端连接器和主题

无论需要保持两个、三个还是更多数据库同步，都适用相同的概念。以下是具有三个数据库的事件图：

每个数据库和表只需要一个主题，因此要捕获SQL Server上客户表上的更改，就需要一个sqlserver.customer主题，该主题可以使用可用的连接器接收任意数据。

总结

现在已经回顾了一个现实中的业务问题，其中Kafka Connect帮助一家用户保持多个应用/数据库同步，以获得更好的客户体验。

良好的应用架构提供了与其他应用共享数据的灵活性，同时支持数据演进。使用AVRO的Kafka Connect解决了这些问题，并提供了各种好处，例如：

使多个数据库保持同步可能很棘手。下表包括可能遇到的一些问题的简明列表，以及这些问题的解决方案。

问题解决方案

数据库使用特定的数据类型，比如date、datetime、smallint、money

使用SMT将此类特定于数据库的数据类型转换为AVRO支持的类型。

两端数据模型有差异

Kafka Connect支持转换，另外还可以在upsert模式下插入数据。

父表应始终在子表之前插入一条记录，以防多个表之间的引用完整性。

这是一个难以解决的问题，有很多方法。理想的方法是为子表引入某种延迟，但此解决方案容易出错。

（编辑：晋中站长网）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!