如何用debezium捕获mysql中的数据，在kafka连接中改变数据捕获和使用jdbc接收器？

Question

我在用debezium捕获mysql中的数据时遇到了问题，将数据捕获并使用kafka连接jdbc接收器将其消费到另一个mysql。

因为debezium对kafka主题产生的模式和有效负载与kafka连接jdbc接收器所期望的模式不兼容。

当jdbc接收器想要使用数据并在另一个mysql中创建记录时，我就会异常。

我该如何解决这个问题？

Answer 1

德贝兹产生的消息结构确实与JDBC接收器所期望的不同。JDBC接收器期望消息中的每个字段与行中的字段相对应，因此消息对应于行的“后置”状态。OTOH，Debezium MySQL连接器执行更改数据捕获，这意味着它所做的不仅仅是包含行的最新状态。具体来说，连接器输出带有包含行的主键或唯一键列的键的消息，以及包含以下信封结构的消息值：

• 操作，例如它是插入、更新还是删除。
• 更改发生前的行状态(插入时为null)
• 更改发生后行的状态(删除时为null)
• 特定于源的信息，包括服务器元数据、事务ID、数据库和表名、事件发生时的服务器时间戳以及事件所在位置的详细信息等。
• 连接器生成事件的时间戳

解决这一差异的最简单方法是使用Kafka 0.10.2.x (目前最新版本为0.10.2.1)和Kafka的新单消息转换(SMT)。每个Kafka连接连接器可以配置零个或多个SMT链，这些链可以在消息写入Kafka之前转换源连接器的输出，或者转换从Kafka读取的消息，然后作为输入传递给接收器连接器。SMT有意非常简单，处理单个消息，绝对不应该访问外部资源或维护任何状态，因此不能替代卡夫卡流或其他更强大的流处理系统，它们可以连接多个输入流，并且可以执行非常复杂的操作并跨多个消息维护状态。

如果您正在使用Kafka流进行任何类型的处理，那么您应该考虑在您的Kafka流应用程序中操作消息结构。如果不是，那么SMT是解决问题的好方法。实际上，有两种方法可以使用SMT来调整消息结构。

第一个选项是使用带有Debezium连接器的SMT来提取/保留行的“后”状态，并在将其写入Kafka之前丢弃所有其他信息。当然，你会在卡夫卡主题中存储更少的信息，并丢弃一些将来可能有价值的疾病预防控制中心信息。

第二个和IMO首选选项是将源连接器保持为-原样，并将所有CDC消息保留在Kafka主题中，但随后使用SMT与接收器连接器一起提取/保留行的“后”状态，并在消息传递到JDBC接收器连接器之前丢弃所有其他信息。您可能可以使用卡夫卡连接中包含的现有SMT之一，但是您可以考虑编写自己的SMT来完成您想做的事情。