13.7. 从 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 记录：

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

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

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

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

与任何 Kafka Connect 连接器配置一样，您可以将 transformations= 设置为多个用逗号分开的 SMT 别名，按您希望 Kafka Connect 应用 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 时，SMT 也在 struct 名称和字段名称之间插入下划线。

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

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

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

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