13.4. 从 Debezium 事件记录中提取字段级别的更改

Debezium 生成具有复杂结构的数据更改事件。每个事件由以下部分组成：

以下示例显示了典型的 Debezium UPDATE 更改事件结构的一部分：

{
	"op": "u",
	"source": {
		...
	},
	"ts_ms" : "...",
	"before" : {
		"field1" : "oldvalue1",
		"field2" : "oldvalue2"
	},
	"after" : {
		"field1" : "newvalue1",
		"field2" : "newvalue2"
	}
}

上例中的消息的复杂格式提供有关源数据库中发生更改的详细信息。但是，格式可能不适用于某些下游用户。sink 连接器或 Kafka 生态系统的其他部分可能会预期信息明确标识数据库操作更改或保持不变的字段。ExtractChangedRecordState SMT 将标头添加到更改事件消息中，以识别由数据库操作修改的字段，以及保持不变的字段。

事件更改 SMT 在 Kafka 记录中从 Debezium UPDATE 更改事件中提取 before 和 after 字段。转换会检查事件状态结构的 before 和 after，以识别操作更改的字段，以及保持不变的字段。根据连接器配置，转换会生成修改的事件消息，添加消息标头来列出更改的字段、更改的字段或两者。如果事件代表 INSERT 或 DELETE，则这个单一消息转换无效。

您可以为 Debezium 连接器配置事件更改 SMT，或使用 Debezium 连接器发送的消息的接收器连接器。如果您希望 Apache Kafka 保留整个原始 Debezium 更改事件，请为接收器连接器配置 SMT。将 SMT 应用到源或接收器连接器的决定取决于您的特定用例。

根据您的用例，您可以通过执行以下任一任务来配置转换来修改原始消息：

您可以通过在连接器的配置中添加 SMT 配置详情，为 Kafka Connect 源或 sink 连接器配置 Debezium 事件更改 SMT。要获取默认行为，不添加任何标头，请将转换添加到连接器配置中，如下例所示：

transforms=changes,...
transforms.changes.type=io.debezium.transforms.ExtractChangedRecordState

与任何 Kafka Connect 连接器配置一样，您可以将 transformations= 设置为多个用逗号分开的 SMT 别名，按您希望 Kafka Connect 应用 SMT 的顺序设置。

以下示例中的连接器配置为事件更改 SMT 设置几个选项：

transforms=changes,...
transforms.changes.type=io.debezium.transforms.ExtractChangedRecordState
transforms.changes.header.changed.name=Changed
transforms.changes.header.unchanged.name=Unchanged

除了 Debezium 连接器在数据库更改时发出的更改事件消息外，连接器还会发出其他类型的信息，包括心跳消息，以及有关 schema 更改和事务的元数据消息。由于这些其他消息的结构与 SMT 设计的更改事件消息的结构不同，因此最好将连接器配置为有选择地应用 SMT，以便它只处理预期的数据更改消息。

有关如何有选择地应用 SMT 的更多信息，请参阅为转换配置 SMT predicate。

下表描述了您可以指定配置事件更改 SMT 的选项。