用于 MongoDB 的 Debezium 连接器

Debezium 的 MongoDB 连接器跟踪 MongoDB 副本集或 MongoDB 分片集群以查找数据库和集合中的文档更改，并将这些更改记录为 Kafka 主题中的事件。连接器自动处理分片集群中分片的添加或删除、每个副本集成员资格的更改、每个副本集中的选举以及等待通信问题的解决。

有关与此连接器兼容的 MongoDB 版本的信息，请参阅Debezium 版本概述。

概述

MongoDB 的复制机制提供了冗余和高可用性，是在生产环境中运行 MongoDB 的首选方式。MongoDB 连接器捕获副本集或分片集群中的更改。

MongoDB副本集由一组服务器组成，这些服务器都具有相同数据的副本，并且复制可确保客户端对副本集主节点上的文档所做的所有更改都正确应用于另一个副本集的服务器，称为辅助节点。MongoDB 复制的工作原理是让主节点在其oplog（或操作日志）中记录更改，然后每个辅助节点读取主节点的 oplog 并按顺序将所有操作应用于自己的文档。将新服务器添加到副本集时，该服务器首先执行快照主节点上的所有数据库和集合，然后读取主节点的 oplog 以应用自快照开始以来可能已进行的所有更改。当这个新服务器赶上主服务器 oplog 的尾部时，它成为辅助服务器（并且能够处理查询）。

MongoDB 连接器支持两种不同的模式来捕获由

capture.mode

选项控制的更改：

• 基于oplog
• 基于更改流

Oplog 捕获模式（旧版）

Debezium MongoDB 连接器使用与上述相同的复制机制，尽管它实际上并没有成为副本集的成员。然而，就像 MongoDB 辅助节点一样，连接器总是读取副本集主节点的 oplog。而且，当连接器第一次看到副本集时，它会查看 oplog 以获取最后记录的事务，然后执行主数据库和集合的快照。复制所有数据后，连接器从之前从 oplog 读取的位置开始流式传输更改。MongoDB oplog 中的操作是幂等的，因此无论操作多少次，它们都会导致相同的结束状态。

这种模式的缺点是只有插入更改事件会包含整个文档，而更新事件只包含更改字段的表示（即无法从更新事件中获取未修改的字段），而删除事件不包含已删除文档的表示除了它的钥匙。

这种模式应该被视为传统模式。MongoDB 5 不支持它，强烈建议用户不要将它用于 MongoDB 4.x 服务器。

更改流模式

Debezium MongoDB 连接器使用与上述类似的复制机制，尽管它实际上并没有成为副本集的成员。主要区别在于连接器不直接读取 oplog，而是将捕获和解码 oplog 委托给 MongoDB 的Change Streams功能。使用更改流，MongoDB 服务器将更改作为事件流公开给集合。Debezium 连接器监视流并将更改传递到下游。而且，当连接器第一次看到副本集时，它会查看 oplog 以获取最后记录的事务，然后执行主数据库和集合的快照。当所有数据都被复制后，连接器然后从它之前从 oplog 读取的位置创建一个更改流。

这是从 MongoDB 4.x 开始的推荐模式。

两种捕获模式都使用存储在偏移量中的不同值，允许它们从连接器重新启动后看到的最后一个位置恢复流式传输。因此，不可能从更改流模式切换到 oplog 模式。为防止任何意外捕获模式更改，连接器具有内置安全检查功能。当连接器启动时，它会检查存储的偏移量。如果原始捕获模式是基于 oplog 的，而新模式是基于变化流的，那么它将尝试迁移到变化流。如果原始捕获模式是基于更改流的，它将继续使用更改流，如果新模式是基于 oplog 的，则将向日志发出关于此的警告。

随着 MongoDB 连接器处理更改，它会定期记录事件起源的 oplog/stream 中的位置。当连接器停止时，它会记录它处理的最后一个 oplog/stream 位置，以便在重新启动时它只是从该位置开始流式传输。换句话说，连接器可以停止、升级或维护，并在一段时间后重新启动，它将准确地从中断的地方重新开始，而不会丢失任何一个事件。当然，MongoDB 的 oplog 通常有一个最大大小的上限，这意味着连接器不应停止太久，否则 oplog 中的某些操作可能会在连接器有机会读取它们之前被清除。在这种情况下，重新启动连接器将检测丢失的 oplog 操作，执行快照，然后继续流式传输更改。

MongoDB 连接器还可以容忍副本集的成员资格和领导力的变化、分片集群中分片的添加或删除以及可能导致通信失败的网络问题。连接器始终使用副本集的主节点流式传输更改，因此当副本集进行选举并且不同的节点成为主节点时，连接器将立即停止流式传输更改，连接到新的主节点，并使用新的主节点开始流式传输更改节点。同样，如果连接器在与主副本集通信时遇到任何问题，它将尝试重新连接（使用指数退避以避免压倒网络或副本集）并从上次中断的地方继续流式传输更改。

其他资源

• 复制机制
• 副本集
• 副本集选举
• 分片集群
• 分片添加
• 碎片清除
• 更改流

MongoDB 连接器的工作原理

连接器支持的 MongoDB 拓扑的概述对于规划您的应用程序很有用。

配置和部署 MongoDB 连接器时，它首先连接到种子地址处的 MongoDB 服务器，并确定每个可用副本集的详细信息。由于每个副本集都有自己独立的 oplog，连接器将尝试为每个副本集使用单独的任务。连接器可以限制它将使用的最大任务数，如果没有足够的任务可用，连接器将为每个任务分配多个副本集，尽管任务仍将为每个副本集使用单独的线程。

对分片集群运行连接器时，使用的值 tasks.max 大于副本集的数量。这将允许连接器为每个副本集创建一个任务，并让 Kafka Connect 跨所有可用工作进程协调、分发和管理任务。

支持的 MongoDB 拓扑

MongoDB 连接器支持以下 MongoDB 拓扑：

• MongoDB副本集

  Debezium MongoDB 连接器可以捕获来自单个MongoDB 副本集的更改。生产副本集至少需要三个成员。要将 MongoDB 连接器与副本集一起使用，请通过连接器的属性提供一个或多个副本集服务器的地址作为种子地址。

  mongodb.hosts

  连接器将使用这些种子连接到副本集，然后一旦连接，将从副本集中获取完整的成员集以及哪个成员是主成员。连接器将启动一个任务以连接到主节点并从主节点的 oplog 中捕获更改。当副本集选出新的主节点时，任务会自动切换到新的主节点。当 MongoDB 以代理为前端时（例如在 OS X 或 Windows 上使用 Docker），当客户端连接到副本集并发现成员时，MongoDB 客户端会将代理排除为有效成员，并尝试并失败直接连接到成员，而不是通过代理。在这种情况下，将连接器的可选

  mongodb.members.auto.discover

  配置属性设置为

  false

  以指示连接器放弃成员资格发现，而是简单地使用第一个种子地址（通过

  mongodb.hosts

  属性指定）作为主节点。这可能有效，但在选举发生时仍然会导致问题。

• MongoDB分片集群

  一个MongoDB 分片集群包括：一个或多个分片，每个分片部署为一个副本集；一个单独的副本集，充当集群的配置服务器客户端连接的一个或多个路由器（也称为

  mongos

  ）并将请求路由到适当的分片要将 MongoDB 连接器与分片集群一起使用，请使用配置服务器副本集的主机地址配置连接器。当连接器连接到这个副本集时，它发现它正在充当分片集群的配置服务器，发现集群中用作分片的每个副本集的信息，然后将启动一个单独的任务来捕获每个副本集的变化。如果将新分片添加到集群或移除现有分片，连接器将自动相应地调整其任务。

• MongoDB独立服务器

  MongoDB 连接器无法监控独立 MongoDB 服务器的更改，因为独立服务器没有 oplog。如果将独立服务器转换为具有一个成员的副本集，则连接器将起作用。

MongoDB 不建议在生产环境中运行独立服务器。有关更多信息，请参阅MongoDB 文档。

逻辑连接器名称

连接器配置属性

mongodb.name

用作MongoDB 副本集或分片集群的*逻辑名称。*连接器以多种方式使用逻辑名称：作为所有主题名称的前缀，以及在记录每个副本集的 oplog/change 流位置时作为唯一标识符。

您应该为每个 MongoDB 连接器指定一个唯一的逻辑名称，以有意义地描述源 MongoDB 系统。我们建议逻辑名称以字母或下划线字符开头，其余字符以字母数字或下划线开头。

执行快照

当任务使用副本集启动时，它使用连接器的逻辑名称和副本集名称来查找描述连接器先前停止读取更改的位置的*偏移量。*如果可以找到偏移量并且它仍然存在于 oplog 中，则任务立即进行流式更改，从记录的偏移量位置开始。

但是，如果没有找到偏移量或 oplog 不再包含该位置，则任务必须首先通过执行快照获取副本集内容的当前状态。此过程首先记录 oplog 的当前位置并将其记录为偏移量（以及表示快照已启动的标志）。然后该任务将继续复制每个集合，生成尽可能多的线程（直到

snapshot.max.threads

配置属性的值）以并行执行此工作。连接器将为它看到的每个文档记录一个单独的读取事件，该读取事件将包含对象的标识符、对象的完整状态和源有关找到对象的 MongoDB 副本集的信息。源信息还将包括一个标志，表示该事件是在快照期间产生的。

此快照将继续，直到它复制了与连接器的过滤器匹配的所有集合。如果连接器在任务的快照完成之前停止，则在重新启动时连接器会再次开始快照。

在连接器执行任何副本集的快照时，尽量避免任务重新分配和重新配置。连接器生成日志消息以报告快照的进度。为了提供最大的控制，请为每个连接器运行一个单独的 Kafka Connect 集群。

即席快照

默认情况下，连接器仅在首次启动后才运行初始快照操作。在这个初始快照之后，在正常情况下，连接器不会重复快照过程。连接器捕获的任何未来更改事件数据仅通过流式处理进入。

但是，在某些情况下，连接器在初始快照期间获得的数据可能会变得陈旧、丢失或不完整。为了提供一种重新捕获收集数据的机制，Debezium 包含一个执行临时快照的选项。数据库中的以下更改可能会导致执行临时快照：

• 修改连接器配置以捕获一组不同的集合。
• Kafka 主题被删除，必须重建。
• 由于配置错误或其他问题而发生数据损坏。

您可以通过启动所谓的ad-hoc 快照为之前捕获快照的集合重新运行快照。即席快照需要使用信号集合。您可以通过向 Debezium 信号集合发送信号请求来启动临时快照。

当您启动现有集合的临时快照时，连接器会将内容附加到集合已存在的主题中。如果删除了以前存在的主题，如果启用了自动主题创建，Debezium 可以自动创建主题。

即席快照信号指定要包含在快照中的集合。快照可以捕获数据库的全部内容，或仅捕获数据库中集合的子集。

execute-snapshot

您可以通过向信令集合发送消息来指定要捕获的集合。将

execute-snapshot

信号的类型设置为

incremental

，并提供要包含在快照中的集合的名称，如下表所述：

场地	默认	价值
type	incremental	指定要运行的快照类型。 设置类型是可选的。目前，您只能请求 incremental 快照。
data-collections	不适用	一个数组，其中包含要生成快照的集合的完全限定名称。名称的格式与配置选项 的格式相同。 signal.data.collection

触发临时快照

execute-snapshot

您可以通过将具有信号类型的条目添加到信令集合来启动临时快照。连接器处理完消息后，将开始快照操作。快照进程读取第一个和最后一个主键值，并将这些值用作每个集合的起点和终点。根据集合中的条目数和配置的块大小，Debezium 将集合划分为块，并继续对每个块进行快照，一次一个。

目前，

execute-snapshot

操作类型仅触发增量快照。有关详细信息，请参阅增量快照。

增量快照

为了提供管理快照的灵活性，Debezium 包含一个补充快照机制，称为增量快照。增量快照依靠 Debezium 机制向 Debezium 连接器发送信号。增量快照基于DDD-3设计文档。

在增量快照中，Debezium 不是像在初始快照中那样一次捕获数据库的完整状态，而是在一系列可配置的块中分阶段捕获每个集合。您可以指定您希望快照捕获的集合以及每个块的大小。块大小决定了快照在数据库上的每次提取操作期间收集的行数。增量快照的默认块大小为 1 KB。

随着增量快照的进行，Debezium 使用水印来跟踪其进度，维护它捕获的每个集合行的记录。与标准初始快照过程相比，这种分阶段捕获数据的方法具有以下优势：

• 您可以在流式数据捕获的同时运行增量快照，而不是将流式传输推迟到快照完成。连接器在整个快照过程中继续从更改日志中捕获近乎实时的事件，并且两个操作都不会阻塞另一个操作。
• 如果增量快照的进度中断，您可以恢复它而不会丢失任何数据。进程恢复后，快照从它停止的点开始，而不是从头重新捕获集合。
• 您可以随时按需运行增量快照，并根据需要重复该过程以适应数据库更新。例如，您可以在修改连接器配置以将集合添加到其

  collection.include.list

  属性后重新运行快照。

增量快照过程

当您运行增量快照时，Debezium 按主键对每个集合进行排序，然后根据配置的块大小将集合拆分为块。逐块工作，然后捕获块中的每个集合行。对于它捕获的每一行，快照都会发出一个

READ

事件。该事件表示块的快照开始时行的值。

随着快照的进行，其他进程可能会继续访问数据库，从而可能会修改集合记录。为反映此类更改，

INSERT

、

UPDATE

或

DELETE

操作将照常提交到事务日志。同样，正在进行的 Debezium 流式处理继续检测这些更改事件并将相应的更改事件记录发送到 Kafka。

Debezium 如何解决具有相同主键的记录之间的冲突

在某些情况下，流式处理发出的

UPDATE

或

DELETE

事件被乱序接收。也就是说，流式处理可能会在快照捕获包含该行的

READ

事件的块之前发出一个修改集合行的事件。当快照最终为该行发出相应的

READ

事件时，它的值已经被取代。为了确保以正确的逻辑顺序处理乱序到达的增量快照事件，Debezium 采用了一种缓冲方案来解决冲突。只有在解决了快照事件和流事件之间的冲突后，Debezium 才会向 Kafka 发出事件记录。

快照窗口

为了帮助解决延迟到达事件和修改同一集合行的流事件之间的冲突

READ

，Debezium 采用了所谓的快照窗口。快照窗口划分了增量快照为指定收集块捕获数据的时间间隔。在一个块的快照窗口打开之前，Debezium 遵循其通常的行为并从事务日志直接向下游发送事件到目标 Kafka 主题。但是从特定块的快照打开的那一刻起，直到它关闭，Debezium 执行重复数据删除步骤以解决具有相同主键的事件之间的冲突。

对于每个数据集合，Debezium 发出两种类型的事件，并将它们的记录存储在单个目标 Kafka 主题中。它直接从表中捕获的快照记录作为

READ

操作发出。同时，随着用户不断更新数据集合中的记录，事务日志也更新以反映每次提交，Debezium 会针对每次更改发出

UPDATE

或操作。

DELETE

当快照窗口打开时，Debezium 开始处理快照块，它将快照记录传递到内存缓冲区。在快照窗口期间，

READ

缓冲区中事件的主键与传入流事件的主键进行比较。如果未找到匹配项，则将流式事件记录直接发送到 Kafka。如果 Debezium 检测到匹配，它会丢弃缓冲的

READ

事件，并将流式记录写入目标主题，因为流式事件在逻辑上取代了静态快照事件。块的快照窗口关闭后，缓冲区仅包含

READ

不存在相关事务日志事件的事件。Debezium 将这些剩余

READ

事件发送到集合的 Kafka 主题。

连接器对每个快照块重复该过程。

触发增量快照

目前，启动增量快照的唯一方法是将临时快照信号发送到源数据库上的信令集合。

INSERT

您将信号作为 SQL查询提交给集合。Debezium 检测到信号集合中的变化后，它会读取信号，并运行请求的快照操作。

您提交的查询指定要包含在快照中的集合，并且可以选择指定快照操作的类型。目前，快照操作的唯一有效选项是默认值

incremental

.

要指定要包含在快照中的集合，请提供一个

data-collections

列出集合的数组，例如，

{"data-collections": ["public.MyFirstTable", "public.MySecondTable"]}

增量快照信号的

data-collections

数组没有默认值。如果

data-collections

数组为空，Debezium 检测到不需要任何操作并且不执行快照。

如果要包含在快照中的集合 . 的名称在数据库、模式或表的名称中包含点 ()，则要将集合添加到 data-collections 数组中，必须用双引号对名称的每个部分进行转义. 例如，要包含存在于 **public** 架构中且名称为 的表 **My.Table** ，请使用以下格式： **"public"."My.Table"** .

先决条件

• 信令已启用。
  • 源数据库上存在信令数据集合，连接器配置为捕获它。
  • 信令数据集合在

    signal.data.collection

    属性中指定。

程序

1. 发送 SQL 查询以将临时增量快照请求添加到信令集合：

   例如，

   id

   命令中的、

   type

   、参数的值

   data

   对应于信令集合的字段。

   以下集合描述了这些参数：

   价值	描述
   myschema.debezium_signal	指定源数据库上的信令集合的完全限定名称
   ad-hoc-1	该 id 参数指定一个任意字符串，该字符串被分配为 id 信号请求的标识符。 使用此字符串将日志消息标识到信令集合中的条目。Debezium 不使用此字符串。相反，在快照期间，Debezium 会生成自己的 id 字符串作为水印信号。
   execute-snapshot	指定 type 参数指定信号要触发的操作。
   data-collections	信号字段的必需组件， data 它指定要包含在快照中的集合名称数组。 该数组按集合的完全限定名称列出集合，使用的格式与您在 signal.data.collection 配置属性中指定连接器的信号集合的名称相同。
   incremental	信号字段的可选 type 组件 data ，指定要运行的快照操作的种类。 目前，唯一有效的选项是默认值 incremental . 在您提交给信号集合的 SQL 查询中指定一个 type 值是可选的。 如果您未指定值，则连接器将运行增量快照。

以下示例显示了连接器捕获的增量快照事件的 JSON。

示例：增量快照事件消息

{
    "before":null,
    "after": {
        "pk":"1",
        "value":"New data"
    },
    "source": {
        ...
        "snapshot":"incremental" 
    },
    "op":"r", 
    "ts_ms":"1620393591654",
    "transaction":null
}
           

物品	字段名称	描述
1	snapshot	指定要运行的快照操作的类型。 目前，唯一有效的选项是默认值 incremental . 在您提交给信号集合的 SQL 查询中指定一个 type 值是可选的。 如果您未指定值，则连接器将运行增量快照。
2	op	指定事件类型。 快照事件的值为 r ，表示 READ 操作。

当前仅对单个副本集部署支持增量快照。此限制将在下一个版本中删除。

流式传输更改

副本集的连接器任务记录偏移量后，它使用偏移量来确定 oplog 中应该开始流式传输更改的位置。然后该任务（取决于配置）连接到副本集的主节点或连接到副本集范围的更改流并从该位置开始流式传输更改。它处理所有创建、插入和删除操作，并将它们转换为 Debezium更改事件。每个更改事件都包括在 oplog 中找到操作的位置，连接器会定期将此记录为最近的偏移量。记录偏移量的时间间隔由 控制

offset.flush.interval.ms

，这是 Kafka Connect 工作器配置属性。

当连接器正常停止时，会记录最后处理的偏移量，以便在重新启动时，连接器将准确地从中断处继续。但是，如果连接器的任务意外终止，那么这些任务可能在它最后一次记录偏移量之后但在记录最后一个偏移量之前已经处理并生成了事件；重新启动时，连接器从最后记录的偏移量开始，可能会生成一些之前在崩溃之前生成的相同事件。

当一切都在名义上运行时，Kafka 消费者实际上只会看到每条消息一次*。但是，当出现问题时，Kafka 只能保证消费者至少会看到每条消息一次*。因此，您的消费者需要预期会多次看到消息。

如上所述，连接器任务总是使用副本集的主节点从 oplog 流式传输更改，确保连接器尽可能看到最新的操作，并且可以以比辅助节点更低的延迟捕获更改改为使用。当副本集选择一个新的主节点时，连接器立即停止流式传输更改，连接到新的主节点，并开始从同一位置的新主节点流式传输更改。同样，如果连接器在与副本集成员通信时遇到任何问题，它会尝试通过使用指数退避来重新连接，以免淹没副本集，并且一旦连接，它就会从上次中断的地方继续流式传输更改。这样，

总而言之，MongoDB 连接器在大多数情况下都会继续运行。通信问题可能会导致连接器等待问题解决。

主题名称

MongoDB 连接器将针对每个集合中的文档的所有插入、更新和删除操作的事件写入单个 Kafka 主题。Kafka 主题的名称始终采用logicalName形式。数据库名称。collectionName，其中logicalName是使用配置属性指定的连接器的逻辑名称， databaseName是发生操作的数据库的名称，collectionName是受影响文档所在的MongoDB 集合的名称。

mongodb.name

例如，考虑一个 MongoDB 副本集，其

inventory

数据库包含四个集合：

products

、

products_on_hand

、

customers

和

orders

。如果监控此数据库的连接器的逻辑名称为

fulfillment

，则连接器将在以下四个 Kafka 主题上产生事件：

• fulfillment.inventory.products

• fulfillment.inventory.products_on_hand

• fulfillment.inventory.customers

• fulfillment.inventory.orders

请注意，主题名称不包含副本集名称或分片名称。因此，对分片集合（每个分片包含集合文档的一个子集）的所有更改都转到同一个 Kafka 主题。

您可以设置 Kafka 以根据需要自动创建主题。如果没有，那么您必须在启动连接器之前使用 Kafka 管理工具来创建主题。

分区

MongoDB 连接器没有明确决定如何为事件划分主题。相反，它允许 Kafka 确定如何根据事件键对主题进行分区。

Partitioner

您可以通过在 Kafka Connect 工作程序配置中定义实现的名称来更改 Kafka 的分区逻辑。

Kafka 仅维护写入单个主题分区的事件的总顺序。按键对事件进行分区确实意味着具有相同键的所有事件总是进入同一个分区。这确保了特定文档的所有事件总是完全有序的。

交易元数据

Debezium 可以生成表示事务元数据边界的事件并丰富更改数据事件消息。

Debezium 接收交易元数据的时间限制Debezium 仅注册和接收部署连接器后发生的事务的元数据。部署连接器之前发生的事务的元数据不可用。

对于每笔交易

BEGIN

和

END

，Debezium 都会生成一个包含以下字段的事件：

• status

  BEGIN

  或者

  END

• id

  唯一交易标识符的字符串表示。

• event_count

  （用于

  END

  活动）

  事务发出的事件总数。

• data_collections

  （用于

  END

  活动）

  一对

  data_collection

  和的数组

  event_count

  ，提供源自给定数据集合的更改所发出的事件数。

以下示例显示了一条典型消息：

{
  "status": "BEGIN",
  "id": "1462833718356672513",
  "event_count": null,
  "data_collections": null
}

{
  "status": "END",
  "id": "1462833718356672513",
  "event_count": 2,
  "data_collections": [
    {
      "data_collection": "rs0.testDB.collectiona",
      "event_count": 1
    },
    {
      "data_collection": "rs0.testDB.collectionb",
      "event_count": 1
    }
  ]
}
           

除非通过

transaction.topic

选项覆盖，否则事务事件将写入名为

*database.server.name*.transaction

.

更改数据事件丰富

启用事务元数据后，数据消息

Envelope

会增加一个新

transaction

字段。此字段以字段组合的形式提供有关每个事件的信息：

• id

  唯一交易标识符的字符串表示。

• total_order

  该事件在事务生成的所有事件中的绝对位置。

• data_collection_order

  事务发出的所有事件中事件的每个数据收集位置。

以下是消息外观的示例：

{
  "patch": null,
  "after": "{\"_id\" : {\"$numberLong\" : \"1004\"},\"first_name\" : \"Anne\",\"last_name\" : \"Kretchmar\",\"email\" : \"[email protected]\"}",
  "source": {
...
  },
  "op": "c",
  "ts_ms": "1580390884335",
  "transaction": {
    "id": "1462833718356672513",
    "total_order": "1",
    "data_collection_order": "1"
  }
}
           

数据更改事件

Debezium MongoDB 连接器为每个插入、更新或删除数据的文档级操作生成数据更改事件。每个事件都包含一个键和一个值。键和值的结构取决于更改的集合。

Debezium 和 Kafka Connect 是围绕连续的事件消息流设计的。但是，这些事件的结构可能会随着时间的推移而发生变化，这对于消费者来说可能难以处理。为了解决这个问题，每个事件都包含其内容的模式，或者，如果您使用的是模式注册表，消费者可以使用它从注册表中获取模式的模式 ID。这使得每个事件都是独立的。

以下骨架 JSON 显示了更改事件的基本四个部分。但是，如何配置您选择在应用程序中使用的 Kafka Connect 转换器决定了这四个部分在更改事件中的表示。仅当您将转换器配置为生成该字段时，该

schema

字段才处于更改事件中。同样，仅当您将转换器配置为生成它时，事件键和事件有效负载才在更改事件中。如果您使用 JSON 转换器并将其配置为生成所有四个基本更改事件部分，则更改事件具有以下结构：

{
 "schema": { 
   ...
  },
 "payload": { 
   ...
 },
 "schema": { 
   ...
 },
 "payload": { 
   ...
 },
}
           

物品	字段名称	描述
1	schema	第一个 schema 字段是事件键的一部分。它指定了一个 Kafka Connect 模式，该模式描述了事件键 payload 部分中的内容。换句话说，第一个 schema 字段描述了已更改文档的键的结构。
2	payload	第一个 payload 字段是事件键的一部分。它具有前一个字段描述的结构， schema 并且包含已更改文档的键。
3	schema	第二个 schema 字段是事件值的一部分。它指定了描述事件值 payload 部分内容的 Kafka Connect 模式。换句话说，第二个 schema 描述了被改变的文档的结构。通常，此模式包含嵌套模式。
4	payload	第二个 payload 字段是事件值的一部分。它具有前一个字段描述的结构， schema 并且包含已更改文档的实际数据。

默认情况下，连接器流将事件记录更改为名称与事件的原始集合相同的主题。请参阅主题名称。

MongoDB 连接器确保所有 Kafka Connect 模式名称都遵循Avro 模式名称格式。这意味着逻辑服务器名称必须以拉丁字母或下划线开头，即 az、AZ 或 _。逻辑服务器名称中的每个剩余字符以及数据库和集合名称中的每个字符都必须是拉丁字母、数字或下划线，即 az、AZ、0-9 或 _。如果存在无效字符，则将其替换为下划线字符。如果逻辑服务器名称、数据库名称或集合名称包含无效字符，并且将名称彼此区分开来的唯一字符无效并因此替换为下划线，这可能会导致意外冲突。

更改事件键

更改事件的键包含更改文档键的模式和更改文档的实际键。对于给定的集合，模式及其相应的有效负载都包含一个

id

字段。此字段的值是文档的标识符，表示为从MongoDB 扩展 JSON 序列化严格模式派生的字符串。

考虑一个逻辑名称为 的连接器、一个包含数据库

fulfillment

的副本集和一个包含如下文档的集合。

inventory``customers

示例文档

{
  "_id": 1004,
  "first_name": "Anne",
  "last_name": "Kretchmar",
  "email": "[email protected]"
}
           

示例更改事件键

捕获对集合的更改的每个更改事件

customers

都具有相同的事件键架构。只要

customers

集合具有前面的定义，捕获

customers

集合更改的每个更改事件都具有以下关键结构。在 JSON 中，它看起来像这样：

{
  "schema": { 
    "type": "struct",
    "name": "fulfillment.inventory.customers.Key", 
    "optional": false, 
    "fields": [ 
      {
        "field": "id",
        "type": "string",
        "optional": false
      }
    ]
  },
  "payload": { 
    "id": "1004"
  }
}
           

物品	字段名称	描述
1	schema	键的模式部分指定了描述键 payload 部分内容的 Kafka Connect 模式。
2	fulfillment.inventory.customers.Key	定义密钥有效负载结构的架构名称。此架构描述了已更改文档的键的结构。键架构名称的格式为connector-name。数据库名称。集合名称。 Key . 在这个例子中： fulfillment 是生成此事件的连接器的名称。 inventory 是包含已更改集合的数据库。 customers 是包含已更新文档的集合。
3	optional	payload 指示事件键是否必须在其字段中包含值。在此示例中，密钥的有效负载中的值是必需的。当文档没有键时，键的有效负载字段中的值是可选的。
4	fields	指定 中预期的 payload 每个字段，包括每个字段的名称、类型以及是否需要。
5	payload	包含为其生成此更改事件的文档的键。在此示例中，键包含一个 id 类型 string 为 的字段 1004 。

此示例使用具有整数标识符的文档，但任何有效的 MongoDB 文档标识符都以相同的方式工作，包括文档标识符。对于文档标识符，事件键的

payload.id

值是一个字符串，它将更新文档的原始

_id

字段表示为使用严格模式的 MongoDB 扩展 JSON 序列化。下表提供了如何

_id

表示不同类型字段的示例。

类型	MongoDB _id 价值	密钥的有效载荷
整数	1234	{ "id" : "1234" }
漂浮	12.34	{ "id" : "12.34" }
细绳	“1234”	{ "id" : "\"1234\"" }
文档	{ "hi" : "kafka", "nums" : [10.0, 100.0, 1000.0] }	{ "id" : "{\"hi\" : \"kafka\", \"nums\" : [10.0, 100.0, 1000.0]}" }
对象标识	ObjectId("596e275826f08b2730779e1f")	{ "id" : "{\"$oid\" : \"596e275826f08b2730779e1f\"}" }
二进制	BinData("a2Fma2E=",0)	{ "id" : "{\"$binary\" : \"a2Fma2E=\", \"$type\" : \"00\"}" }

更改事件值

更改事件中的值比键复杂一些。与键一样，值也有一个

schema

部分和一个

payload

部分。该

schema

部分包含描述该部分

Envelope

结构的架构

payload

，包括其嵌套字段。创建、更新或删除数据的操作的更改事件都有一个带有信封结构的值负载。

考虑用于显示更改事件键示例的相同示例文档：

示例文档

{
  "_id": 1004,
  "first_name": "Anne",
  "last_name": "Kretchmar",
  "email": "[email protected]"
}
           

针对每个事件类型描述了更改此文档的更改事件的值部分：

• 创建事件
• 更新事件
• 删除事件
• 墓碑事件

创建事件

以下示例显示了连接器为在

customers

集合中创建数据的操作生成的更改事件的值部分：

{
    "schema": { 
      "type": "struct",
      "fields": [
        {
          "type": "string",
          "optional": true,
          "name": "io.debezium.data.Json", 
          "version": 1,
          "field": "after"
        },
        {
          "type": "string",
          "optional": true,
          "name": "io.debezium.data.Json",
          "version": 1,
          "field": "patch"
        },
        {
          "type": "struct",
          "fields": [
            {
              "type": "string",
              "optional": false,
              "field": "version"
            },
            {
              "type": "string",
              "optional": false,
              "field": "connector"
            },
            {
              "type": "string",
              "optional": false,
              "field": "name"
            },
            {
              "type": "int64",
              "optional": false,
              "field": "ts_ms"
            },
            {
              "type": "boolean",
              "optional": true,
              "default": false,
              "field": "snapshot"
            },
            {
              "type": "string",
              "optional": false,
              "field": "db"
            },
            {
              "type": "string",
              "optional": false,
              "field": "rs"
            },
            {
              "type": "string",
              "optional": false,
              "field": "collection"
            },
            {
              "type": "int32",
              "optional": false,
              "field": "ord"
            },
            {
              "type": "int64",
              "optional": true,
              "field": "h"
            }
          ],
          "optional": false,
          "name": "io.debezium.connector.mongo.Source", 
          "field": "source"
        },
        {
          "type": "string",
          "optional": true,
          "field": "op"
        },
        {
          "type": "int64",
          "optional": true,
          "field": "ts_ms"
        }
      ],
      "optional": false,
      "name": "dbserver1.inventory.customers.Envelope" 
      },
    "payload": { 
      "after": "{\"_id\" : {\"$numberLong\" : \"1004\"},\"first_name\" : \"Anne\",\"last_name\" : \"Kretchmar\",\"email\" : \"[email protected]\"}", 
      "patch": null,
      "source": { 
        "version": "1.9.5.Final",
        "connector": "mongodb",
        "name": "fulfillment",
        "ts_ms": 1558965508000,
        "snapshot": false,
        "db": "inventory",
        "rs": "rs0",
        "collection": "customers",
        "ord": 31,
        "h": 1546547425148721999
      },
      "op": "c", 
      "ts_ms": 1558965515240 
    }
  }
           

物品	字段名称	描述
1	schema	值的架构，它描述了值的有效负载的结构。在连接器为特定集合生成的每个更改事件中，更改事件的值架构都是相同的。
2	name	在该 schema 部分中，每个 name 字段都为值的有效负载中的字段指定架构。 io.debezium.data.Json 是有效负载的 after 、 patch 和 filter 字段的架构。此架构特定于 customers 集合。创建事件是唯一一种包含 after 字段的事件。一个更新事件包含一个 filter 字段和一个 patch 字段。删除事件包含一个字段 filter ，但不包含 after 字段或 patch 字段。
3	name	io.debezium.connector.mongo.Source 是有效负载 source 字段的架构。此模式特定于 MongoDB 连接器。连接器将它用于它生成的所有事件。
4	name	dbserver1.inventory.customers.Envelope 是有效负载整体结构的架构，其中 dbserver1 是连接器名称， inventory 是数据库， customers 是集合。此架构特定于集合。
5	payload	该值的实际数据。这是更改事件提供的信息。 看起来事件的 JSON 表示比它们描述的文档大得多。这是因为 JSON 表示必须包括消息的模式和有效负载部分。但是，通过使用Avro 转换器，您可以显着减小连接器流式传输到 Kafka 主题的消息的大小。
6	after	一个可选字段，指定事件发生后文档的状态。在此示例中，该 after 字段包含新文档的 _id 、 first_name 、 last_name 和 email 字段的值。该 after 值始终是一个字符串。按照惯例，它包含文档的 JSON 表示。MongoDB的OPLOG条目仅包含用于_Create_事件的文档的完整状态，以及设置为选项 update 时的事件；in other words, a create event is the only kind of event that contains an after field, when the option is set either to or . capture.mode``change_streams_update_full``capture.mode``oplog``change_streams
7	source	描述事件源元数据的必填字段。此字段包含可用于将此事件与其他事件进行比较的信息，包括事件的来源、事件发生的顺序以及事件是否属于同一事务的一部分。源元数据包括：Debezium 版本。生成事件的连接器的名称。MongoDB 副本集的逻辑名称，形成生成事件的命名空间，并用于连接器写入的 Kafka 主题名称。包含新文档的集合和数据库的名称。如果事件是快照的一部分。在数据库中进行更改时的时间戳以及时间戳内事件的序号。MongoDB操作的唯一标识，取决于MongoDB的版本。它要么是 h oplog 事件中的字段，要么是一个名为 的字段 stxnid ，它表示来自 oplog 事件的 lsid 和 txnNumber 字段（仅限 oplog 捕获模式）。MongoDB 会话的唯一标识符 lsid 和事务编号 txnNumber ，以防更改在事务内部执行（仅限更改流捕获模式）。
8	op	强制字符串，描述导致连接器生成事件的操作类型。在本例中， c 表示该操作创建了一个文档。有效值为： c = 创建 u = 更新 d = 删除 r = 读取（仅适用于快照）
9	ts_ms	显示连接器处理事件的时间的可选字段。该时间基于运行 Kafka Connect 任务的 JVM 中的系统时钟。 在 source 对象中， ts_ms 指示在数据库中进行更改的时间。通过比较 for payload.source.ts_ms 的值 payload.ts_ms ，您可以确定源数据库更新和 Debezium 之间的延迟。

更新事件

Oplog 捕获模式（旧版）

示例集合中更新的更改事件的值与该集合的创建

customers

事件具有相同的模式。同样，事件值的有效负载具有相同的结构。但是，事件值有效负载在更新事件中包含不同的值。更新事件没有值。相反，它具有以下两个字段：

after

• patch

  是一个字符串字段，包含幂等更新操作的 JSON 表示

• filter

  是一个字符串字段，其中包含更新选择条件的 JSON 表示形式。该

  filter

  字符串可以包含多个分片集合的分片键字段。

customers

以下是连接器为集合中的更新生成的事件中的更改事件值示例：

{
    "schema": { ... },
    "payload": {
      "op": "u", 
      "ts_ms": 1465491461815, 
      "patch": "{\"$set\":{\"first_name\":\"Anne Marie\"}}", 
      "filter": "{\"_id\" : {\"$numberLong\" : \"1004\"}}", 
      "source": { 
        "version": "1.9.5.Final",
        "connector": "mongodb",
        "name": "fulfillment",
        "ts_ms": 1558965508000,
        "snapshot": false,
        "db": "inventory",
        "rs": "rs0",
        "collection": "customers",
        "ord": 6,
        "h": 1546547425148721999
      }
    }
  }
           

物品	字段名称	描述
1	op	强制字符串，描述导致连接器生成事件的操作类型。在此示例中， u 表示操作更新了文档。
2	ts_ms	显示连接器处理事件的时间的可选字段。该时间基于运行 Kafka Connect 任务的 JVM 中的系统时钟。 在 source 对象中， ts_ms 指示在数据库中进行更改的时间。通过比较 for payload.source.ts_ms 的值 payload.ts_ms ，您可以确定源数据库更新和 Debezium 之间的延迟。
3	patch	包含对文档的实际 MongoDB 幂等更改的 JSON 字符串表示。在此示例中，更新将 first_name 字段更改为新值。 更新事件值不包含字段 after 。
4	filter	包含用于标识要更新的文档的 MongoDB 选择标准的 JSON 字符串表示。
5	source	描述事件源元数据的必填字段。此字段包含与同一集合的创建事件相同的信息，但值不同，因为此事件来自 oplog 中的不同位置。源元数据包括：Debezium 版本。生成事件的连接器的名称。MongoDB 副本集的逻辑名称，形成生成事件的命名空间，并用于连接器写入的 Kafka 主题名称。包含更新文档的集合和数据库的名称。如果事件是快照的一部分。在数据库中进行更改时的时间戳以及时间戳内事件的序号。MongoDB操作的唯一标识，取决于MongoDB的版本。它要么是 h oplog 事件中的字段，要么是一个名为 的字段 stxnid ，它表示来自 oplog 事件的 lsid 和 txnNumber 字段。

在 Debezium 更改事件中，MongoDB 提供 patch 字段的内容。该字段的格式取决于 MongoDB 数据库的版本。因此，当您升级到较新的 MongoDB 数据库版本时，请为格式可能发生的变化做好准备。本文档中的示例来自 MongoDB 3.4，在您的应用程序中，事件格式可能会有所不同。

在 MongoDB 的 oplog 中，更新事件不包含更改文档的前后状态**。因此，Debezium 连接器无法提供此信息。但是，Debezium 连接器在创建和读取事件中提供文档的起始状态。流的下游消费者可以通过保持每个文档的最新状态并将新事件中的状态与保存的状态进行比较来重建文档状态。Debezium 连接器无法保持此状态。

更改流捕获模式

示例集合中更新的更改事件的值与该集合的创建

customers

事件具有相同的模式。同样，事件值的有效负载具有相同的结构。但是，事件值有效负载在更新事件中包含不同的值。更新事件仅在设置为选项时才具有值。在这种情况下，有一个带有一些附加字段的新结构化字段：

after``capture.mode``change_streams_update_full``updateDescription

• updatedFields

  是一个字符串字段，其中包含更新的文档字段的 JSON 表示及其值

• removedFields

  是从文档中删除的字段名称列表

• truncatedArrays

  是文档中被截断的数组列表

customers

以下是连接器为集合中的更新生成的事件中的更改事件值示例：

{
    "schema": { ... },
    "payload": {
      "op": "u", 
      "ts_ms": 1465491461815, 
      "after":"{\"_id\": {\"$numberLong\": \"1004\"},\"first_name\": \"Anne Marie\",\"last_name\": \"Kretchmar\",\"email\": \"[email protected]\"}", 
      "updateDescription": {
        "removedFields": null,
        "updatedFields": "{\"first_name\": \"Anne Marie\"}", 
        "truncatedArrays": null
      },
      "source": { 
        "version": "1.9.5.Final",
        "connector": "mongodb",
        "name": "fulfillment",
        "ts_ms": 1558965508000,
        "snapshot": false,
        "db": "inventory",
        "rs": "rs0",
        "collection": "customers",
        "ord": 1,
        "h": null,
        "tord": null,
        "stxnid": null,
        "lsid":"{\"id\": {\"$binary\": \"FA7YEzXgQXSX9OxmzllH2w==\",\"$type\": \"04\"},\"uid\": {\"$binary\": \"47DEQpj8HBSa+/TImW+5JCeuQeRkm5NMpJWZG3hSuFU=\",\"$type\": \"00\"}}",
        "txnNumber":1
      }
    }
  }
           

物品	字段名称	描述
1	op	强制字符串，描述导致连接器生成事件的操作类型。在此示例中， u 表示操作更新了文档。
2	ts_ms	显示连接器处理事件的时间的可选字段。该时间基于运行 Kafka Connect 任务的 JVM 中的系统时钟。 在 source 对象中， ts_ms 指示在数据库中进行更改的时间。通过比较 for payload.source.ts_ms 的值 payload.ts_ms ，您可以确定源数据库更新和 Debezium 之间的延迟。
3	after	包含实际 MongoDB 文档的 JSON 字符串表示。 如果捕获模式未设置为，则更新事件值不包含字段 after``change_streams_update_full
4	updatedFields	包含文档的更新字段值的 JSON 字符串表示形式。在此示例中，更新将 first_name 字段更改为新值。
5	source	描述事件源元数据的必填字段。此字段包含与同一集合的创建事件相同的信息，但值不同，因为此事件来自 oplog 中的不同位置。源元数据包括：Debezium 版本。生成事件的连接器的名称。MongoDB 副本集的逻辑名称，形成生成事件的命名空间，并用于连接器写入的 Kafka 主题名称。包含更新文档的集合和数据库的名称。如果事件是快照的一部分。在数据库中进行更改时的时间戳以及时间戳内事件的序号。 lsid MongoDB 会话和事务号的唯一标识符， txnNumber 以防更改在事务内执行。

事件中的 after 值应作为文档的时间点值处理。该值不是动态计算的，而是从集合中获取的。因此，如果多个更新一个接一个地紧随其后，则所有更新更新事件将包含相同的 after 值，该值将表示存储在文档中的最后一个值。如果您的应用程序依赖于渐变进化，那么您应该 updateDescription 只依赖。

删除事件

删除更改事件中的值与同一集合的创建和更新

schema

事件具有相同的部分。删除事件中的部分包含与同一集合的创建和更新事件不同的值。特别是，删除事件既不包含值也不包含一个或多个值。以下是集合中文档的删除事件示例：

payload``after``patch``updateDescription``customers

{
    "schema": { ... },
    "payload": {
      "op": "d", 
      "ts_ms": 1465495462115, 
      "filter": "{\"_id\" : {\"$numberLong\" : \"1004\"}}", 
      "source": { 
        "version": "1.9.5.Final",
        "connector": "mongodb",
        "name": "fulfillment",
        "ts_ms": 1558965508000,
        "snapshot": true,
        "db": "inventory",
        "rs": "rs0",
        "collection": "customers",
        "ord": 6,
        "h": 1546547425148721999
      }
    }
  }
           

物品	字段名称	描述
1	op	描述操作类型的强制字符串。 op 字段值为 d ，表示该文档已被删除。
2	ts_ms	显示连接器处理事件的时间的可选字段。该时间基于运行 Kafka Connect 任务的 JVM 中的系统时钟。 在 source 对象中， ts_ms 指示在数据库中进行更改的时间。通过比较 for payload.source.ts_ms 的值 payload.ts_ms ，您可以确定源数据库更新和 Debezium 之间的延迟。
3	filter	包含用于标识要删除的文档的 MongoDB 选择标准的 JSON 字符串表示形式（仅限 oplog 捕获模式）。
4	source	描述事件源元数据的必填字段。此字段包含与同一集合的创建或更新事件相同的信息，但值不同，因为此事件来自 oplog 中的不同位置。源元数据包括：Debezium 版本。生成事件的连接器的名称。MongoDB 副本集的逻辑名称，形成生成事件的命名空间，并用于连接器写入的 Kafka 主题名称。包含已删除文档的集合和数据库的名称。如果事件是快照的一部分。在数据库中进行更改时的时间戳以及时间戳内事件的序号。MongoDB操作的唯一标识，取决于MongoDB的版本。它要么是 h oplog 事件中的字段，要么是一个名为 的字段 stxnid ，它表示来自 oplog 事件的 lsid 和 txnNumber 字段（仅限 oplog 捕获模式）。MongoDB 会话的唯一标识符 lsid 和事务编号 txnNumber ，以防更改在事务内部执行（仅限更改流捕获模式）。

MongoDB 连接器事件旨在与Kafka 日志压缩一起使用。只要至少保留每个键的最新消息，日志压缩就可以删除一些较旧的消息。这让 Kafka 可以回收存储空间，同时确保主题包含完整的数据集并可用于重新加载基于键的状态。

墓碑事件

唯一标识的文档的所有 MongoDB 连接器事件都具有完全相同的键。删除文档时，删除事件值仍然适用于日志压缩，因为 Kafka 可以删除所有具有相同键的早期消息。但是，要让 Kafka 删除所有具有该键的消息，消息值必须是

null

. 为了实现这一点，在 Debezium 的 MongoDB 连接器发出删除事件后，连接器会发出一个特殊的墓碑事件，该事件具有相同的键但有一个

null

值。墓碑事件通知 Kafka 可以删除具有相同密钥的所有消息。

设置 MongoDB

MongoDB 连接器使用 MongoDB 的 oplog/change 流来捕获更改，因此连接器仅适用于 MongoDB 副本集或分片集群，其中每个分片都是一个单独的副本集。有关设置副本集或分片集群的信息，请参阅 MongoDB 文档。此外，请务必了解如何使用副本集启用访问控制和身份验证。

您还必须有一个具有适当角色的 MongoDB 用户来读取

admin

可以读取 oplog 的数据库。此外，用户还必须能够读取分片

config

集群的配置服务器中的数据库，并且必须具有

listDatabases

特权操作。当使用更改流（默认）时，用户还必须具有集群范围的权限操作

find

和

changeStream

.

云中的 MongoDB

您可以将 MongoDB 的 Debezium 连接器与MongoDB Atlas一起使用。将 Debezium 连接到 MongoDB Atlas 时，启用其中之一

capture modes

基于更改流，而不是 oplog。请注意，MongoDB Atlas 仅支持通过 SSL 的安全连接，即

+mongodb.ssl.enabled

连接器选项必须设置为

true

.

部署

要部署 Debezium MongoDB 连接器，请安装 Debezium MongoDB 连接器存档，配置连接器，然后通过将其配置添加到 Kafka Connect 来启动连接器。

先决条件

• 已安装Apache Zookeeper、Apache Kafka和Kafka Connect。
• MongoDB 已安装并设置为与 Debezium 连接器一起使用。

程序

1. 下载 连接器的插件存档，
2. 将 JAR 文件提取到您的 Kafka Connect 环境中。
3. 将包含 JAR 文件的目录添加到Kafka Connect 的

   plugin.path

   .
4. 重新启动 Kafka Connect 进程以获取新的 JAR 文件。

如果您正在使用不可变容器，请参阅Debezium 的Apache Zookeeper、Apache Kafka 和 Kafka Connect 容器映像，其中 MongoDB 连接器已安装并准备好运行。

您还可以在 Kubernetes 和 OpenShift 上运行 Debezium。

Debezium教程将引导您使用这些图像，这是了解 Debezium 的好方法。

MongoDB 连接器配置示例

以下是连接器实例的配置示例，该实例从位于

rs0

192.168.99.100 的端口 27017 的 MongoDB 副本集中捕获数据，我们在逻辑上将其命名为

fullfillment

. 通常，您通过设置可用于连接器的配置属性，在 JSON 文件中配置 Debezium MongoDB 连接器。

您可以选择为特定的 MongoDB 副本集或分片集群生成事件。或者，您可以过滤掉不需要的集合。

{
  "name": "inventory-connector", 
  "config": {
    "connector.class": "io.debezium.connector.mongodb.MongoDbConnector", 
    "mongodb.hosts": "rs0/192.168.99.100:27017", 
    "mongodb.name": "fullfillment", 
    "collection.include.list": "inventory[.]*" 
  }
}
           

当我们向 Kafka Connect 服务注册连接器时的名称。
MongoDB 连接器类的名称。
用于连接到 MongoDB 副本集的主机地址。
MongoDB 副本集的逻辑名称，形成生成事件的命名空间，并用于连接器写入的所有 Kafka 主题的名称、Kafka Connect 模式名称以及 Avro 时相应 Avro 模式的命名空间使用转换器。
与要监视的所有集合的集合命名空间（例如，.）匹配的正则表达式列表。这是可选的。

有关可以为 Debezium MongoDB 连接器设置的配置属性的完整列表，请参阅MongoDB 连接器配置属性。

您可以使用命令将此配置发送

POST

到正在运行的 Kafka Connect 服务。该服务记录配置并启动一个执行以下操作的连接器任务：

• 连接到 MongoDB 副本集或分片集群。
• 为每个副本集分配任务。
• 如有必要，执行快照。
• 读取 oplog/change 流。
• 流将事件记录更改为 Kafka 主题。

添加连接器配置

要开始运行 Debezium MongoDB 连接器，请创建一个连接器配置，并将该配置添加到您的 Kafka Connect 集群。

先决条件

• MongoDB 设置为使用 Debezium 连接器。
• Debezium MongoDB 连接器已安装。

程序

1. 为 MongoDB 连接器创建配置。
2. 使用Kafka Connect REST API将该连接器配置添加到您的 Kafka Connect 集群。

结果

连接器启动后，它会完成以下操作：

• 对 MongoDB 副本集中的集合执行一致的快照。
• 读取副本集的 oplogs/change 流。
• 为每个插入、更新和删除的文档生成更改事件。
• 流将事件记录更改为 Kafka 主题。

连接器属性

Debezium MongoDB 连接器具有许多配置属性，您可以使用这些属性为您的应用程序实现正确的连接器行为。许多属性都有默认值。有关属性的信息组织如下：

• 必需的 Debezium MongoDB 连接器配置属性
• 高级 Debezium MongoDB 连接器配置属性

除非默认值可用，否则需要以下配置属性。

财产	默认	描述
name	连接器的唯一名称。尝试使用相同名称再次注册将失败。（所有 Kafka Connect 连接器都需要此属性。）
connector.class	连接器的 Java 类的名称。始终 io.debezium.connector.mongodb.MongoDbConnector 为 MongoDB 连接器使用值。
mongodb.hosts	副本集中 MongoDB 服务器的主机名和端口对的逗号分隔列表（以“主机”或“主机：端口”的形式）。该列表可以包含单个主机名和端口对。如果 mongodb.members.auto.discover 设置为 false ，则主机和端口对应以副本集名称为前缀（例如， rs0/localhost:27017 ）。+必须提供当前的主地址。此限制将在下一个 Debezium 版本中删除。
mongodb.name	标识此连接器监控的连接器和/或 MongoDB 副本集或分片集群的唯一名称。每台服务器最多应该由一个 Debezium 连接器监控，因为该服务器名称作为所有持久 Kafka 主题的前缀，这些主题来自 MongoDB 副本集或集群。仅使用字母数字字符、连字符、点和下划线来构成名称。逻辑名称在所有其他连接器中应该是唯一的，因为该名称用作命名从该连接器接收记录的 Kafka 主题的前缀。+不要更改此属性的值。如果您更改名称值，在重新启动后，连接器不会继续向原始主题发出事件，而是向名称基于新值的主题发出后续事件。
mongodb.user	连接到 MongoDB 时要使用的数据库用户的名称。仅当 MongoDB 配置为使用身份验证时才需要这样做。
mongodb.password	连接 MongoDB 时使用的密码。仅当 MongoDB 配置为使用身份验证时才需要这样做。
mongodb.authsource	admin	包含 MongoDB 凭据的数据库（身份验证源）。仅当 MongoDB 配置为对另一个身份验证数据库使用身份验证而不是 admin .
mongodb.ssl.enabled	false	连接器将使用 SSL 连接到 MongoDB 实例。
mongodb.ssl.invalid.hostname.allowed	false	启用 SSL 时，此设置控制在连接阶段是否禁用严格的主机名检查。如果 true 连接不会阻止中间人攻击。
database.include.list	空字符串	一个可选的逗号分隔的正则表达式列表，匹配要监控的数据库名称；任何未包含在其中的数据库名称都将 database.include.list 被排除在监控之外。默认情况下，所有数据库都受到监视。不得与 database.exclude.list .
database.exclude.list	空字符串	一个可选的以逗号分隔的正则表达式列表，匹配要从监控中排除的数据库名称；任何未包含在其中的数据库名称 database.exclude.list 都会受到监控。不得与 database.include.list .
collection.include.list	空字符串	一个可选的以逗号分隔的正则表达式列表，匹配要监控的 MongoDB 集合的完全限定命名空间；任何未包含在其中的集合都将 collection.include.list 被排除在监控之外。每个标识符的格式为databaseName。集合名称。默认情况下，连接器将监视除 local 和 admin 数据库中的所有集合之外的所有集合。不得与 collection.exclude.list .
collection.exclude.list	空字符串	一个可选的以逗号分隔的正则表达式列表，它与要从监控中排除的 MongoDB 集合的完全限定名称空间匹配；任何未包含在其中的集合 collection.exclude.list 都会受到监控。每个标识符的格式为databaseName。集合名称。不得与 collection.include.list .
snapshot.mode	initial	指定在连接器启动时运行快照的条件。默认值为initial，并指定连接器在未找到偏移量或 oplog/change 流不再包含先前偏移量时读取快照。never选项指定连接器不应该使用快照，而是连接器应该继续跟踪日志。
capture.mode	change_streams_update_full	指定用于从 MongoDB 服务器捕获更改的方法。默认值为change_streams_update_full，并指定连接器通过 MongoDB 更改流机制捕获更改，并且更新事件应包含完整文档。change_streams模式将使用相同的捕获方法，但更新事件不会包含完整的文档。oplog模式指定直接访问MongoDB oplog ；这是传统方法，不应用于新的连接器实例。
snapshot.include.collection.list	中指定的所有集合 collection.include.list	一个可选的、以逗号分隔的正则表达式列表，与 collection.include.list 您要为其拍摄快照的指定模式的名称相匹配。
field.exclude.list	空字符串	应从更改事件消息值中排除的字段的完全限定名称的可选逗号分隔列表。字段的完全限定名称的格式为databaseName。集合名称。字段名。nestedFieldName，其中databaseName和collectionName可能包含匹配任何字符的通配符 (*)。
field.renames	空字符串	应用于重命名更改事件消息值中的字段的字段的完全限定替换的可选逗号分隔列表。字段的完全限定替换形式为databaseName。集合名称。字段名。nestedFieldName : newNestedFieldName，其中databaseName和collectionName可以包含匹配任何字符的通配符 (*)，冒号 (😃 用于确定字段的重命名映射。下一个字段替换应用于列表中上一个字段替换的结果，因此在重命名同一路径中的多个字段时请记住这一点。
tasks.max	1	应为此连接器创建的最大任务数。MongoDB 连接器将尝试为每个副本集使用单独的任务，因此在将连接器与单个 MongoDB 副本集一起使用时，默认值是可以接受的。将连接器与 MongoDB 分片集群一起使用时，我们建议指定一个等于或大于集群中分片数量的值，以便每个副本集的工作可以由 Kafka Connect 分配。
snapshot.max.threads	1	正整数值，指定用于执行副本集中集合的初始同步的最大线程数。默认为 1。
tombstones.on.delete	true	控制删除事件后是否有墓碑事件。 true - 删除操作由删除事件和随后的墓碑事件表示。 false - 只发出一个删除事件。 删除源记录后，发出 tombstone 事件（默认行为）允许 Kafka 完全删除与已删除行的键相关的所有事件，以防主题启用日志压缩。
snapshot.delay.ms	连接器在启动后拍摄快照之前应等待的时间间隔（以毫秒为单位）； 可用于在集群中启动多个连接器时避免快照中断，这可能导致连接器重新平衡。
snapshot.fetch.size		指定在拍摄快照时应从每个集合中一次性读取的最大文档数。连接器将分批读取该大小的集合内容。 默认为 0，表示服务器选择了合适的 fetch 大小。
schema.name.adjustment.mode	欧元	指定应如何调整架构名称以与连接器使用的消息转换器兼容。可能的设置： avro 用下划线替换不能在 Avro 类型名称中使用的字符。 none 不应用任何调整。

以下高级配置属性具有良好的默认值，适用于大多数情况，因此很少需要在连接器的配置中指定。

财产	默认	描述
max.batch.size	2048	正整数值，指定在此连接器的每次迭代期间应处理的每批事件的最大大小。默认为 2048。
max.queue.size	8192	正整数值，指定阻塞队列可以保存的最大记录数。当 Debezium 读取从数据库流式传输的事件时，它会将事件放入阻塞队列中，然后再将它们写入 Kafka。在连接器接收消息的速度快于将消息写入 Kafka 的速度或 Kafka 不可用时，阻塞队列可以为从数据库读取更改事件提供背压。当连接器定期记录偏移量时，将忽略队列中保存的事件。始终将 的值设置 max.queue.size 为大于 的值 max.batch.size 。
max.queue.size.in.bytes		一个长整数值，指定阻塞队列的最大容量（以字节为单位）。默认情况下，没有为阻塞队列指定卷限制。要指定队列可以使用的字节数，请将此属性设置为正长值。 如果 max.queue.size 也设置了，当队列的大小达到任一属性指定的限制时，写入队列将被阻止。例如，如果设置 max.queue.size=1000 , 和 max.queue.size.in.bytes=5000 ，则在队列包含 1000 条记录或队列中的记录量达到 5000 字节后阻止写入队列。
poll.interval.ms	1000	正整数值，指定连接器在每次迭代期间应等待新更改事件出现的毫秒数。默认为 1000 毫秒或 1 秒。
connect.backoff.initial.delay.ms	1000	正整数值，指定在第一次连接尝试失败或没有可用的主节点后尝试重新连接主节点时的初始延迟。默认为 1 秒（1000 毫秒）。
connect.backoff.max.delay.ms	1000	正整数值，指定在重复连接尝试失败或没有可用的主节点后尝试重新连接主节点时的最大延迟。默认为 120 秒（120,000 毫秒）。
connect.max.attempts	16	正整数值，指定在发生异常和任务被中止之前与副本集主节点的最大失败连接尝试次数。默认为 16，默认为 16， connect.backoff.initial.delay.ms 导致 connect.backoff.max.delay.ms 在失败前尝试超过 20 分钟。
mongodb.members.auto.discover	true	布尔值，指定 ‘mongodb.hosts’ 中的地址是否是应用于发现集群或副本集的所有成员的种子（ true ），或者是否 mongodb.hosts 应按原样使用（ false ）中的地址。默认值是 true 并且应该在所有情况下使用，除了 MongoDB前面有代理的情况。
source.struct.version	v2	source CDC 事件中块的架构版本。Debezium 0.10 对块的结构进行了一些重大更改， source 以统一所有连接器的暴露结构。 通过将此选项设置为 v1 可以生成早期版本中使用的结构。请注意，不建议使用此设置，并计划在未来的 Debezium 版本中删除此设置。
heartbeat.interval.ms		控制发送心跳消息的频率。 此属性包含以毫秒为单位的时间间隔，该时间间隔定义连接器将消息发送到心跳主题的频率。这可用于监视连接器是否仍在从数据库接收更改事件。如果在较长时间内仅更改未捕获集合中的记录，您还应该利用心跳消息。在这种情况下，连接器将继续从数据库中读取 oplog/change 流，但永远不会向 Kafka 发出任何更改消息，这反过来意味着没有偏移量更新提交给 Kafka。这将导致 oplog 文件被轮换，但连接器不会注意到它，因此在重新启动时某些事件不再可用，这导致需要重新执行初始快照。将此参数设置 为根本不发送心跳消息。 默认禁用。
heartbeat.topics.prefix	__debezium-heartbeat	控制向其发送心跳消息的主题的命名。 主题是根据模式命名的 <heartbeat.topics.prefix>.<server.name> 。
sanitize.field.names	true 当连接器配置明确指定使用 Avro 的 key.converter 或 value.converter 参数时，默认为 false .	字段名称是否经过清理以符合 Avro 命名要求。有关详细信息，请参阅Avro 命名。
skipped.operations	流式传输期间将跳过的操作类型的逗号分隔列表。操作包括： c 插入/创建、 u 更新和 d 删除。默认情况下，不会跳过任何操作。
snapshot.collection.filter.overrides	控制哪些集合项包含在快照中。此属性仅影响快照。以databaseName.collectionName形式指定以逗号分隔的集合名称列表。对于您指定的每个集合，还要指定另一个配置属性：. 例如，其他配置属性的名称可能是：. 将此属性设置为仅检索快照中所需的项目的有效过滤器表达式。当连接器执行快照时，它只检索与过滤器表达式匹配的项目。 snapshot.collection.filter.overrides.*databaseName*.*collectionName*``snapshot.collection.filter.overrides.customers.orders
provide.transaction.metadata	false	当设置为 true Debezium 时，会生成带有事务边界的事件，并使用事务元数据丰富数据事件信封。有关其他详细信息，请参阅事务元数据。
transaction.topic	${database.server.name}.transaction	控制连接器向其发送事务元数据消息的主题的名称。占位符 ${database.server.name} 可用于引用连接器的逻辑名称（请参阅逻辑连接器名称）；默认为 ${database.server.name}.transaction ，例如 dbserver1.transaction 。
retriable.restart.connector.wait.ms	10000（10 秒）	发生可重试错误后重新启动连接器之前等待的毫秒数。
mongodb.poll.interval.ms	30000	连接器轮询新的、删除的或更改的副本集的时间间隔。
mongodb.connect.timeout.ms	10000（10 秒）	在中止新的连接尝试之前驱动程序将等待的毫秒数。
mongodb.socket.timeout.ms	在发生超时之前，套接字上的发送/接收可能需要的毫秒数。值 禁用此行为。
mongodb.server.selection.timeout.ms	30000（30 秒）	驱动程序在超时并引发错误之前等待选择服务器的毫秒数。
cursor.max.await.time.ms		指定 oplog/change 流游标在导致执行超时异常之前等待服务器产生结果的最大毫秒数。值 指示使用服务器/驱动程序默认等待超时。
signal.data.collection	无默认值	用于向连接器发送信号的数据集合的完全限定名称。使用以下格式指定集合名称： *<databaseName>*.*<collectionName>*
incremental.snapshot.chunk.size	1024	连接器在增量快照块期间获取并读入内存的最大文档数。增加块大小提供了更高的效率，因为快照运行的快照查询更少，但更大的大小。然而，更大的块大小也需要更多的内存来缓冲快照数据。将块大小调整为在您的环境中提供最佳性能的值。

监控

除了 Zookeeper、Kafka 和 Kafka Connect 拥有的对 JMX 指标的内置支持之外，Debezium MongoDB 连接器还有两种指标类型。

• 快照指标在执行快照时提供有关连接器操作的信息。
• 当连接器捕获更改和流式处理更改事件记录时，流式处理指标提供有关连接器操作的信息。

Debezium监控文档提供了有关如何使用 JMX 公开这些指标的详细信息。

快照指标

MBean是. _

debezium.mongodb:type=connector-metrics,context=snapshot,server=*<mongodb.server.name>*,task=*<task.id>*

除非快照操作处于活动状态，或者自上次连接器启动以来已发生快照，否则不会公开快照指标。

下表列出了可用的快照指标。

属性	类型	描述
LastEvent	string	连接器读取的最后一个快照事件。
MilliSecondsSinceLastEvent	long	自连接器读取并处理最新事件以来的毫秒数。
TotalNumberOfEventsSeen	long	自上次启动或重置以来，此连接器已看到的事件总数。
NumberOfEventsFiltered	long	已被连接器上配置的包含/排除列表过滤规则过滤的事件数。
CapturedTables	string[]	连接器捕获的表列表。
QueueTotalCapacity	int	用于在快照程序和主 Kafka Connect 循环之间传递事件的队列长度。
QueueRemainingCapacity	int	用于在快照程序和主 Kafka Connect 循环之间传递事件的队列的可用容量。
TotalTableCount	int	快照中包含的表总数。
RemainingTableCount	int	快照尚未复制的表数。
SnapshotRunning	boolean	快照是否已启动。
SnapshotAborted	boolean	快照是否已中止。
SnapshotCompleted	boolean	快照是否完成。
SnapshotDurationInSeconds	long	到目前为止，快照已拍摄的总秒数，即使未完成也是如此。
RowsScanned	Map<String, Long>	包含为快照中的每个表扫描的行数的映射。在处理过程中，表格会逐渐添加到地图中。每扫描 10,000 行并在完成表时更新。
MaxQueueSizeInBytes	long	队列的最大缓冲区（以字节为单位）。 max.queue.size.in.bytes 如果设置为正长值，则此指标可用。
CurrentQueueSizeInBytes	long	队列中的当前记录量（以字节为单位）。

Debezium MongoDB 连接器还提供以下自定义快照指标：

属性	类型	描述
NumberOfDisconnects	long	数据库断开连接的次数。

流媒体指标

MBean是. _

debezium.mongodb:type=connector-metrics,context=streaming,server=*<mongodb.server.name>*,task=*<task.id>*

下表列出了可用的流式指标。

属性	类型	描述
LastEvent	string	连接器读取的最后一个流事件。
MilliSecondsSinceLastEvent	long	自连接器读取并处理最新事件以来的毫秒数。
TotalNumberOfEventsSeen	long	自上次启动或指标重置以来，此连接器已看到的事件总数。
TotalNumberOfCreateEventsSeen	long	自上次启动或指标重置以来，此连接器已看到的创建事件总数。
TotalNumberOfUpdateEventsSeen	long	自上次启动或指标重置以来，此连接器已看到的更新事件总数。
TotalNumberOfDeleteEventsSeen	long	自上次启动或指标重置以来，此连接器已看到的删除事件总数。
NumberOfEventsFiltered	long	已被连接器上配置的包含/排除列表过滤规则过滤的事件数。
CapturedTables	string[]	连接器捕获的表列表。
QueueTotalCapacity	int	用于在流媒体和主 Kafka Connect 循环之间传递事件的队列长度。
QueueRemainingCapacity	int	队列的空闲容量，用于在流媒体和 Kafka Connect 主循环之间传递事件。
Connected	boolean	表示连接器当前是否连接到数据库服务器的标志。
MilliSecondsBehindSource	long	上次更改事件的时间戳与处理它的连接器之间的毫秒数。这些值将包含运行数据库服务器和连接器的机器上的时钟之间的任何差异。
NumberOfCommittedTransactions	long	已提交的已处理事务数。
SourceEventPosition	Map<String, String>	最后接收到的事件的坐标。
LastTransactionId	string	最后处理的事务的事务标识符。
MaxQueueSizeInBytes	long	队列的最大缓冲区（以字节为单位）。 max.queue.size.in.bytes 如果设置为正长值，则此指标可用。
CurrentQueueSizeInBytes	long	队列中的当前记录量（以字节为单位）。

Debezium MongoDB 连接器还提供以下自定义流指标：

属性	类型	描述
NumberOfDisconnects	long	数据库断开连接的次数。
NumberOfPrimaryElections	long	主节点选举次数。

MongoDB 连接器常见问题

Debezium 是一个分布式系统，可以捕获多个上游数据库中的所有更改，并且永远不会错过或丢失任何事件。当系统正常运行并得到仔细管理时，Debezium 会为每个更改事件提供一次交付。

如果发生故障，系统不会丢失任何事件。但是，当它从故障中恢复时，它可能会重复一些更改事件。在这种情况下，像 Kafka 一样，Debezium至少提供一次更改事件的传递。

本节的其余部分描述了 Debezium 如何处理各种故障和问题。

配置和启动错误

在以下情况下，连接器尝试启动失败，在日志中报告错误或异常，并停止运行：

• 连接器的配置无效。
• 连接器无法使用指定的连接参数成功连接到 MongoDB。

失败后，连接器会尝试使用指数退避重新连接。您可以配置重新连接尝试的最大次数。

在这些情况下，错误将包含有关问题的更多详细信息以及可能的建议解决方法。更正配置或解决 MongoDB 问题后，可以重新启动连接器。

MongoDB 变得不可用

一旦连接器运行，如果任何 MongoDB 副本集的主节点变得不可用或无法访问，连接器将反复尝试重新连接到主节点，使用指数退避来防止网络或服务器饱和。如果在可配置的连接尝试次数后主节点仍然不可用，则连接器将失败。

重新连接的尝试由三个属性控制：

• connect.backoff.initial.delay.ms

  - 第一次尝试重新连接之前的延迟，默认为 1 秒（1000 毫秒）。

• connect.backoff.max.delay.ms

  - 尝试重新连接之前的最大延迟，默认为 120 秒（120,000 毫秒）。

• connect.max.attempts

  - 产生错误前的最大尝试次数，默认为 16。

每个延迟是先前延迟的两倍，直到最大延迟。在给定默认值的情况下，下表显示了每次失败的连接尝试的延迟以及失败前的总累积时间。

重新连接尝试次数	尝试前的延迟，以秒为单位	尝试前的总延迟，以分钟和秒为单位
1	1	00:01
2	2	00:03
3	4	00:07
4	8	00:15
5	16	00:31
6	32	01:03
7	64	02:07
8	120	04:07
9	120	06:07
10	120	08:07
11	120	10:07
12	120	12:07
13	120	14:07
14	120	16:07
15	120	18:07
16	120	20:07

Kafka Connect 进程正常停止

如果 Kafka Connect 正在分布式模式下运行，并且 Kafka Connect 进程正常停止，那么在关闭该进程之前，Kafka Connect 会将所有进程的连接器任务迁移到该组中的另一个 Kafka Connect 进程，并且新的连接器任务将准确地从先前任务停止的地方开始。在连接器任务正常停止并在新进程上重新启动时，处理会有短暂的延迟。

如果组仅包含一个进程并且该进程正常停止，则 Kafka Connect 将停止连接器并记录每个副本集的最后偏移量。重新启动后，副本集任务将完全从中断的地方继续。

Kafka Connect 进程崩溃

如果 Kafka 连接器进程意外停止，那么它正在运行的任何连接器任务都将终止，而不会记录它们最近处理的偏移量。当 Kafka Connect 在分布式模式下运行时，它将在其他进程上重新启动这些连接器任务。但是，MongoDB 连接器将从早期进程记录的最后一个偏移量恢复，这意味着新的替换任务可能会生成一些在崩溃之前处理的相同更改事件。重复事件的数量取决于偏移刷新周期和崩溃前的数据量变化。

因为在故障恢复期间某些事件可能会重复，所以消费者应该始终预期某些事件可能会重复。Debezium 的变化是幂等的，所以一系列事件总是导致相同的状态。Debezium 还在每个更改事件消息中包含有关事件起源的源特定信息，包括 MongoDB 事件的唯一事务标识符 ( h ) 和时间戳 ( sec 和 ord )。消费者可以跟踪这些值中的其他值，以了解它是否已经看到特定事件。

卡夫卡变得不可用

当连接器生成更改事件时，Kafka Connect 框架使用 Kafka 生产者 API 在 Kafka 中记录这些事件。Kafka Connect 还将按照您在 Kafka Connect 工作程序配置中指定的频率定期记录这些更改事件中出现的最新偏移量。如果 Kafka 代理变得不可用，运行连接器的 Kafka Connect 工作进程将简单地重复尝试重新连接到 Kafka 代理。换句话说，连接器任务将简单地暂停，直到可以重新建立连接，此时连接器将准确地从中断的地方恢复。

连接器长时间停止

如果连接器正常停止，则可以继续使用副本集，并且任何新的更改都会记录在 MongoDB 的 oplog 中。当连接器重新启动时，它将为上次停止的每个副本集恢复流式更改，记录连接器停止时所做的所有更改的更改事件。如果连接器停止的时间足够长，以至于 MongoDB 从其 oplog 中清除了一些连接器尚未读取的操作，那么在启动时连接器将执行快照。

正确配置的 Kafka 集群能够提供巨大的吞吐量。Kafka Connect 是使用 Kafka 最佳实践编写的，如果有足够的资源，也将能够处理非常大量的数据库更改事件。正因为如此，当连接器在一段时间后重新启动时，它很可能会赶上数据库，尽管速度将取决于 Kafka 的功能和性能以及对 MongoDB 中的数据所做的更改量。

如果连接器保持停止的时间足够长，MongoDB 可能会清除旧的 oplog 文件，并且连接器的最后位置可能会丢失。在这种情况下，当配置了初始快照模式（默认）的连接器最终重新启动时，MongoDB 服务器将不再具有起始点，连接器将失败并报错。

MongoDB 丢失写入

在某些故障情况下，MongoDB 可能会丢失提交，从而导致 MongoDB 连接器无法捕获丢失的更改。例如，如果主节点在应用更改并将更改记录到其 oplog 后突然崩溃，则 oplog 可能在辅助节点读取其内容之前变得不可用。因此，被选为新主节点的辅助节点可能会丢失其 oplog 中的最新更改。

目前，在 MongoDB 中没有办法防止这种副作用。