$changeStreamSplitLargeEvent (聚合)

本页内容

定义

行为
示例

定义

$changeStreamSplitLargeEvent

从 MongoDB 7.0（和 6.0.9）开始。

如果当更改流包含超过 16 MB 的大型事件时，会返回一个 BSONObjectTooLarge 异常。从 MongoDB 7.0（和 6.0.9）开始，您可以使用 $changeStreamSplitLargeEvent 阶段将事件拆分为较小的片段。

只有当绝对必要时应使用 $changeStreamSplitLargeEvent。例如，如果您的应用程序需要完整文档的前映像或后映像，并生成超过 16 MB 的大型事件，请使用 $changeStreamSplitLargeEvent。

在您决定使用 $changeStreamSplitLargeEvent 之前，应首先尝试减小更改事件的大小。例如

除非您的应用程序需要，否则不要请求文档的前映像或后映像。这会在更多情况下生成 fullDocument 和 fullDocumentBeforeChange 字段，这些通常是更改事件中最大的对象。
使用 $project 阶段仅包含您应用程序所需的字段。这可以减小更改事件的大小，并避免将大型事件拆分为片段所需的时间。这允许在每批中返回更多更改事件。

您只能在管道中有一个 $changeStreamSplitLargeEvent 阶段，并且它必须是最后一个阶段。您只能在 $changeStream 管道中使用 $changeStreamSplitLargeEvent。

$changeStreamSplitLargeEvent 语法

{
  $changeStreamSplitLargeEvent: {}
}

行为

$changeStreamSplitLargeEvent 将超过16 MB的事件拆分为片段，并使用更改流光标依次返回这些片段。

这些片段被拆分，以便第一个片段返回尽可能多的字段。这确保了事件上下文尽快返回。

当更改事件被拆分时，仅使用顶层字段的长度。 $changeStreamSplitLargeEvent 不会递归处理或拆分子文档。例如，如果您使用 $project 阶段创建一个具有单个字段的事件，该字段大小为20 MB，则事件不会被拆分，该阶段会返回错误。

每个片段都有一个恢复令牌。使用片段的令牌恢复的流将

从下一个片段开始新的流。
如果在序列中的最后一个片段恢复，则从下一个事件开始。

每个事件的一个片段包括一个 splitEvent 文档

splitEvent: {
   fragment: <int>,
   of: <int>
}

以下表格描述了字段。

字段	描述
`fragment`	片段索引，从1开始。
`of`	事件的片段总数。

示例

本节中的示例场景展示了使用新集合 myCollection 的 $changeStreamSplitLargeEvent。

创建 myCollection 并插入一个接近16 MB数据的文档

db.myCollection.insertOne(
   { _id: 0, largeField: "a".repeat( 16 * 1024 * 1024 - 1024 ) }
)

largeField 包含重复的字母 a。

为 myCollection 启用 changeStreamPreAndPostImages，允许更改流检索更新前（预映像）和更新后（后映像）的文档

db.runCommand( {
   collMod: "myCollection",
   changeStreamPreAndPostImages: { enabled: true }
} )

使用 db.collection.watch() 创建更改流光标以监视 myCollection 的更改：

myChangeStreamCursor = db.myCollection.watch(
   [ { $changeStreamSplitLargeEvent: {} } ],
   { fullDocument: "required", fullDocumentBeforeChange: "required" }
)

对于更改流事件

fullDocument: "required" 包含后映像文档。
fullDocumentBeforeChange: "required" 包含文档的原始图像。

有关详细信息，请参阅 $changeStream。

更新 myCollection 中的文档，这也会生成包含文档原始图像和后续图像的变更流事件

db.myCollection.updateOne(
   { _id: 0 },
   { $set: { largeField: "b".repeat( 16 * 1024 * 1024 - 1024 ) } }
)

largeField 现在包含重复的字母 b。

使用 next() 方法从 myChangeStreamCursor 中检索片段，并将片段存储在名为 firstFragment、secondFragment 和 thirdFragment 的对象中

const firstFragment = myChangeStreamCursor.next()
const secondFragment = myChangeStreamCursor.next()
const thirdFragment = myChangeStreamCursor.next()

显示 firstFragment.splitEvent

firstFragment.splitEvent

输出片段的详细信息

splitEvent: { fragment: 1, of: 3 }

同样，secondFragment.splitEvent 和 thirdFragment.splitEvent 返回

splitEvent: { fragment: 2, of: 3 }
splitEvent: { fragment: 3, of: 3 }

要检查 firstFragment 的对象键

Object.keys( firstFragment )

输出

[
   '_id',
   'splitEvent',
   'wallTime',
   'clusterTime',
   'operationType',
   'documentKey',
   'ns',
   'fullDocument'
]

要检查 firstFragment.fullDocument 的字节数大小

bsonsize( firstFragment.fullDocument )

输出

16776223

secondFragment 包含 fullDocumentBeforeChange 原始图像，大小约为 16 MB。以下示例显示了 secondFragment 的对象键

Object.keys( secondFragment )

输出

[ '_id', 'splitEvent', 'fullDocumentBeforeChange' ]

thirdFragment 包含 updateDescription 字段，大小约为 16 MB。以下示例显示了 thirdFragment 的对象键

Object.keys( thirdFragment )

输出

[ '_id', 'splitEvent', 'updateDescription' ]

有关变更流和事件的更多信息，请参阅变更事件。

$changeStream

$collStats