13.8. 在 Debezium MongoDB 更改事件状态后提取源文档

Debezium MongoDB 连接器生成具有复杂结构的更改事件。每个事件信息包括以下部分：

以下片段显示了在 MongoDB 插入 操作后连接器发出的 create change 事件的基本结构：

{
  "op": "c",
  "after": "{\"field1\":\"newvalue1\",\"field2\":\"newvalue1\"}",
  "source": { ... }
}

上例中的 after 字段的复杂格式提供有关源数据库中更改的详细信息。但是，有些使用者无法处理包含嵌套值的消息。要将原始消息的复杂嵌套字段转换为更简单、更通用的兼容结构，请为 MongoDB 使用事件扁平化 SMT。SMT 扁平化消息中嵌套字段的结构，如下例所示：

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

有关 Debezium MongoDB 连接器生成的默认消息结构的更多信息，请参阅 连接器文档。

MongoDB 的事件扁平化 SMT 会从 Debezium MongoDB 连接器发送的 create 或 update 更改事件消息中拉取 after 字段。在 SMT 处理原始更改事件消息后，它会生成一个简化的版本，该版本仅包含 after 字段的内容。

根据您的用例，您可以将 ExtractNewDocumentState SMT 应用到 Debezium MongoDB 连接器，或应用到消耗 Debezium 连接器生成的消息的接收器连接器。如果您将 SMT 应用到 Debezium MongoDB 连接器，则 SMT 会修改连接器在发送到 Apache Kafka 之前发送的消息。为确保 Kafka 以原始格式保留完整的 Debezium 更改事件信息，请将 SMT 应用到接收器连接器。

当您使用事件扁平化 SMT 处理从 MongoDB 连接器发送的消息时，SMT 会将原始消息中的记录结构转换为正确输入的 Kafka Connect 记录，这些记录可以被典型的 sink 连接器使用。例如，SMT 将原始消息中的 after 信息转换为任何消费者可以处理的模式结构的 JSON 字符串。

另外，您可以为 MongoDB 配置事件扁平化 SMT，以便在处理过程中以其他方式修改消息。如需更多信息，请参阅配置 主题。

为 MongoDB 配置事件扁平化(ExtractNewDocumentState) SMT，用于使用 Debezium MongoDB 连接器发送的消息。

以下主题提供详情：

transforms=unwrap,...
transforms.unwrap.type=io.debezium.connector.mongodb.transforms.ExtractNewDocumentState

transforms=unwrap,...
transforms.unwrap.type=io.debezium.connector.mongodb.transforms.ExtractNewDocumentState
transforms.unwrap.drop.tombstones=false
transforms.unwrap.delete.handling.mode=drop
transforms.unwrap.add.headers=op

默认情况下，事件扁平化 SMT 将 MongoDB 阵列转换为与 Apache Kafka Connect 或 Apache Avro 模式兼容的数组。虽然 MongoDB 数组可以包含多个类型元素，但 Kafka 数组中的所有元素都必须是相同的类型。

要确保 SMT 对数组的满足环境需求的方式进行编码，您可以指定 array.encoding 配置选项。以下示例显示了设置数组编码的配置：

transforms=unwrap,...
transforms.unwrap.type=io.debezium.connector.mongodb.transforms.ExtractNewDocumentState
transforms.unwrap.array.encoding=<array|document>

根据配置，SMT 通过使用以下编码方法之一处理源消息中数组的每个实例：

以下示例显示了 Debezium MongoDB 连接器如何代表一个数据库文档，其中包含一个包含异构数据类型的数组：

{
    "_id": 1,
    "a1": [
        {
            "a": 1,
            "b": "none"
        },
        {
            "a": "c",
            "d": "something"
        }
    ]
}

{
    "_id": 1,
    "a1": {
        "_0": {
            "a": 1,
            "b": "none"
        },
        "_1": {
            "a": "c",
            "d": "something"
        }
    }
}

文档 编码选项可让 SMT 处理由异构元素组成的任意数组。但是，在使用这个选项前，请始终验证 sink 连接器和其他下游用户是否可以处理包含多个数据类型的数组。

当数据库操作涉及嵌入式文档时，Debezium MongoDB 连接器会发出一个 Kafka 事件记录，其结构反映了原始文档的分级结构。也就是说，事件消息将嵌套文档表示为一组嵌套字段结构。在下游连接器无法处理包含嵌套结构的消息的环境中，您可以将事件扁平化 SMT 配置为在消息中的扁平化分级结构。扁平消息结构更适合类似表格的存储。

要将 SMT 配置为扁平化嵌套结构，请将 flatten.struct 配置选项设置为 true。在转换的消息中，字段名称被构建为与文档源一致。SMT 通过将父文档字段的名称与嵌套文档字段的名称连接来重命名每个扁平化字段。flatten.struct.delimiter 选项定义的分隔符会将名称的组件分开来。struct.delimiter 的默认值是一个下划线字符(_)。

以下示例显示了指定 SMT flattens 嵌套结构的配置：

transforms=unwrap,...
transforms.unwrap.type=io.debezium.connector.mongodb.transforms.ExtractNewDocumentState
transforms.unwrap.flatten.struct=<true|false>
transforms.unwrap.flatten.struct.delimiter=<string>

以下示例显示了由 MongoDB 连接器发出的事件消息。消息包括了一个文档 a 的字段，其中包含两个嵌套文档 (b 和 c) 的字段：

{
    "_id": 1,
    "a": {
            "b": 1,
            "c": "none"
    },
    "d": 100
}

以下示例中的消息显示了 MongoDB flattens 在以上消息中嵌套结构的 SMT 后的输出：

{
    "_id": 1,
    "a_b": 1,
    "a_c": "none",
    "d": 100
}

在生成的消息中，原始消息中嵌套的 b 和 c 字段将被扁平化并重命名。重命名的字段将父文档 a 的名称与嵌套文档的名称连接在一起： a_b 和 a_c 。新字段名称的组件通过下划线字符分隔，具体由 struct.delimiter 配置属性的定义。

在 MongoDB 中，$unset operator 和 $rename operator 都从文档中删除字段。因为 MongoDB 集合是无模式的，因此在更新从文档中删除了字段后，无法推断更新文档中缺失字段的名称。为了支持接收器连接器或其他可能需要有关删除字段信息的消费者，Debezium 会发出更新消息，其中包含一个 removedFields 元素，其中列出了已删除字段的名称。

以下示例显示一个操作的一个更新消息的部分，这导致删除了 a 字段：

"payload": {
  "op": "u",
  "ts_ms": "...",
  "before": "{ ... }",
  "after": "{ ... }",
  "updateDescription": {
    "removedFields": ["a"],
    "updatedFields": null,
    "truncatedArrays": null
  }
}

在上例中，before 和 after 代表文档更新以前和以后的源文档的状态。只有在为连接器设置了 capture.mode 时，连接器才会发出这些字段，如以下列表中所述：

假设配置为捕获完整文档的连接器，当 ExtractNewDocumentState SMT 收到 $unset 事件 的更新 消息时，通过代表删除的字段具有 null 值来重新编码消息，如下例所示：

{
    "id": 1,
    "a": null
}

对于没有配置为捕获完整文档的连接器，当 SMT 收到 $unset 操作的 update 事件时，它会生成以下输出信息：

{
   "a": null
}

在 SMT flattens 事件消息后，生成的消息不再指示生成事件是 create,update 或 initial snapshot read 的操作。通常，您可以通过将连接器配置为公开与 删除 附带的 tombstone 或 rewrite 事件的信息来识别删除操作。有关将连接器配置为公开有关 tombstones 并重写事件信息的更多信息，请参阅 drop.tombstones 和 delete.handling.mode 属性。

要在事件消息中报告数据库操作类型，SMT 可以在以下元素之一中添加 op 字段：

例如，要添加一个标头属性来显示原始操作的类型，添加转换，然后将 add.headers 属性添加到连接器配置中，如下例所示：

transforms=unwrap,...
transforms.unwrap.type=io.debezium.connector.mongodb.transforms.ExtractNewDocumentState
transforms.unwrap.add.headers=op

根据前面的配置，SMT 通过向消息添加 op 标头来报告事件类型，并为它分配一个字符串值来标识操作类型。分配的字符串值基于原始 MongoDB 更改事件消息 中的 op 字段值。

MongoDB 的事件扁平化 SMT 可将原始更改事件消息中的 metadata 字段添加到简化的消息中。添加的元数据字段的前缀为双下划线("__")。在事件记录中添加元数据可以包括内容，如发生更改事件的集合名称，或包含特定于连接器的字段，如副本集名称。目前，SMT 只能从以下更改事件子结构中添加字段：source, transaction 和 updateDescription。

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

例如，您可以指定以下配置，将副本集名称(rs)和更改事件的集合名称添加到最终扁平化事件记录中：

transforms=unwrap,...
transforms.unwrap.type=io.debezium.connector.mongodb.transforms.ExtractNewDocumentState
transforms.unwrap.add.fields=rs,collection

前面的配置会导致以下内容添加到扁平化的记录中：

{ "__rs" : "rs0", "__collection" : "my-collection", ... }

如果您希望 SMT 添加 metadata 字段 来删除 事件，请将 delete.handling.mode 选项的值设置为 rewrite。

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

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

下表描述了 MongoDB 事件扁平化 SMT 的配置选项。

version:VERSION, connector:CONNECTOR, source.ts_ms:EVENT_TIMESTAMP

version:VERSION, connector:CONNECTOR, source.ts_ms:EVENT_TIMESTAMP