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

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

以下片段显示了 创建 更改事件的基本结构，连接器在 MongoDB insert 操作后发出：

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

上例中的 after 字段复杂格式提供有关源数据库中发生的更改的详细信息。但是，一些消费者无法处理包含嵌套值的消息。要将原始消息的复杂嵌套字段转换为更简单、更通用的兼容结构，请将事件扁平化 SMT 用于 MongoDB。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 记录，这些记录可以被典型的接收器连接器使用。例如，SMT 将原始消息中代表 后 信息的 JSON 字符串转换为任何消费者可以处理的模式结构。

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

为使用 Debezium MongoDB 连接器发出的消息的接收器连接器配置 event flattening (ExtractNewDocumentState) SMT。

以下主题提供详情：

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 处理由异构元素组成的任意数组。但是，在使用这个选项前，请始终验证接收器连接器和其他下游用户是否能够处理包含多个数据类型的数组。

当数据库操作涉及嵌入式文档时，Debebe 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 的 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": "...",
  "ts_us" : "...",
  "ts_ns" : "...",
  "before": "{ ... }",
  "after": "{ ... }",
  "updateDescription": {
    "removedFields": ["a"],
    "updatedFields": null,
    "truncatedArrays": null
  }
}

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

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

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

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

{
   "a": null
}

在 SMT 扁平化事件消息后，生成的消息不再指示生成事件的操作是 创建、更新 或初始快照 读取。通常，您可以通过将连接器配置为公开有关 删除 附带的 tombstone 或重写事件的信息，来识别删除操作。有关配置连接器以公开事件消息中的 tombstones 和 rewrites 的信息的更多信息，请参阅 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 可以将原始更改事件消息的元数据字段添加到简化的消息。添加的元数据字段以双下划线("__")作为前缀。在事件记录中添加元数据可以包括内容，如发生更改事件的集合名称，或者包含特定于连接器的字段，如副本集名称。目前，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 添加元数据字段来删除事件，请将 delete. handling.mode 选项的值设置为 重写。

除了 Debezium 连接器在数据库更改时发出的更改事件消息外，连接器还会发出其他类型的信息，包括 heartbeat 信息，以及有关 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