12.6. 从 Debezium MongoDB 更改事件中提取源文档

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

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

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

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

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

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

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

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

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

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

为使用 Debezium MongoDB 连接器发送的消息的 MongoDB 配置事件扁平化(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 处理由异构元素组成的任意数组。但是，在使用这个选项前，请始终验证接收器连接器和其他下游用户是否能够处理包含多个数据类型的数组。

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

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

以下示例显示了用于指定 SMT 扁平化嵌套结构的配置：

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

以下示例显示了 MongoDB 连接器发送的事件信息。消息包含一个文档的字段，其中包含两个嵌套文档( 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 _b 和 a_c。新字段名称的组件使用下划线字符分隔，具体由 struct.delimiter 配置属性的设置所定义。

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

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

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

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

假设配置为捕获完整文档的连接器，当 ExtractNewDocumentState SMT 收到 $unset 事件 的更新 消息时，对于 $unset 事件的 SMT 会重新处理消息，这代表 removed 字段有一个 null 值，如下例所示：

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

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

{
   "a": null
}

在 SMT 扁平化事件后，生成的消息不再指示生成事件的操作是 create、update 或 initial snapshot read。通常，您可以通过配置连接器来公开有关删除附带的 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 连接器在发生数据库更改时发出的更改事件消息外，连接器还会发出其他类型的消息，包括心跳消息以及有关架构更改和事务的元数据消息。由于这些消息的结构与 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