12.5. 从 Debezium 更改事件中提取源记录 after 状态

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

以下示例显示了 UPDATE 更改事件的消息结构的一部分：

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

有关连接器更改事件结构的更多信息，请参阅连接器的文档。

事件扁平化 SMT 处理上例中的信息后，它简化了消息格式，如下例所示：

{
	"field1" : "newvalue1",
	"field2" : "newvalue2"
}

事件扁平化 SMT 从 Kafka 记录中的 Debezium 更改事件中提取 after 字段。SMT 仅将原始更改事件替换为其 after 字段来创建简单的 Kafka 记录。

您可以为 Debezium 连接器或消耗 Debezium 连接器发送的消息配置事件扁平化 SMT。为接收器连接器配置事件扁平化的优点是，存储在 Apache Kafka 中的记录包含整个 Debezium 更改事件。将 SMT 应用到源或接收器连接器的决定取决于您的特定用例。

您可以配置转换以执行以下操作之一：

数据库 DELETE 操作会导致 Debezium 生成两个 Kafka 记录：

您可以将事件扁平化配置为执行以下操作之一，而不是丢弃包含 before 行数据的记录：

同样，您可以配置事件扁平化 SMT，而不是丢弃 tombstone 记录，以将 tombstone 记录保留在流中。

通过在连接器配置中添加 SMT 配置详情，在 Kafka Connect 源或接收器连接器中配置 Debezium 事件扁平化 SMT。例如，要获取转换的默认行为，请将其添加到连接器配置中，而不指定任何选项，如下例所示：

transforms=unwrap,...
transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState

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

以下 .properties 示例设置了几个事件扁平化 SMT 选项：

transforms=unwrap,...
transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState
transforms.unwrap.drop.tombstones=false
transforms.unwrap.delete.handling.mode=rewrite
transforms.unwrap.add.fields=table,lsn

您可以配置事件扁平化 SMT，将原始更改事件元数据添加到简化的 Kafka 记录中。例如，您可能希望简化的记录的标头或值包含以下任意一种：

要在简化的 Kafka 记录的标头中添加元数据，请指定 add.headers 选项。要在简化的 Kafka 记录的值中添加元数据，请指定 add.fields 选项。每个选项都使用以逗号分隔的更改事件字段名称列表。不要指定空格。当有重复的字段名称时，要为其中一个字段添加元数据，请指定 struct 和字段。例如：

transforms=unwrap,...
transforms.unwrap.type=io.debezium.transforms.ExtractNewRecordState
transforms.unwrap.add.fields=op,table,lsn,source.ts_ms
transforms.unwrap.add.headers=db
transforms.unwrap.delete.handling.mode=rewrite

使用这个配置，简化的 Kafka 记录将包含以下内容：

{
 ...
	"__op" : "c",
	"__table": "MY_TABLE",
	"__lsn": "123456789",
	"__source_ts_ms" : "123456789",
 ...
}

另外，简化的 Kafka 记录会有一个 __db 标头。

在简化的 Kafka 记录中，SMT 使用双下划线为 metadata 字段名称添加前缀。当您指定 struct 时，F SMT 也会在 struct 名称和字段名称之间插入下划线。

要在用于 DELETE 操作的简化 Kafka 记录中添加元数据，还必须配置 delete.handling.mode=rewrite。

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

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

下表描述了您可以指定配置事件扁平化 SMT 的选项。