流处理的聚合管道阶段

Atlas Stream Processing通过处理连续数据流的阶段扩展了聚合管道。这些阶段与默认mongod进程内置的现有聚合阶段相结合，使您能够对连续数据执行许多与对静态数据执行相同的操作。 Atlas Stream Processing管道定义不能超过 16 MB。

下表列出了 Atlas Stream Processing 特有的聚合管道阶段，以及以修改后的形式存在于 Atlas Stream Processing 中的聚合管道阶段。

注意

有关特定操作符（包括事务语法和示例）的详细信息，请单击该操作符的参考页面链接。

聚合管道阶段	用途
$source	指定要从中使用消息的流数据源。
$validate	根据用户定义的模式验证流的文档。
$https	支持在流处理器中对任意 REST API 终结点进行类似 `curl` 的调用。
$lookup	对指定集合执行左外连接，以筛选“已连接”集合中的文档进行处理。此版本的现有 `$lookup` 阶段要求您在连接注册表中指定Atlas集合作为`from`字段的值。
$ skippingWindow	将文档从流分配到具有用户定义的持续时间和开始时间间隔的窗口。
$tumblingWindow	将文档从流分配到具有用户定义的持续时间的不重叠的连续窗口。
$sessionWindow	在输入流中对每个活动“会话”运行管道。每个文档会根据其分区和时间戳被分配到一个会话窗口。
$setStreamMeta	在来自输入流的文档上设置元数据字段。此元数据字段可用于执行动态操作，例如在不更改文档正文内容的情况下对文档执行接收器路由。
$cachedLookup	使用带生存时间的缓存执行查找，以减少对联接集合的重复读取。
$externalFunction	调用管道中的 AWS Lambda函数，可以选择将结果传递到下游阶段。
$throttle	限制流处理器将数据传递到下游阶段的速率，以每秒字节或消息为单位。
$emit	指定连接注册表中要向其发出消息的流或时间序列集合。
$merge	现有`$merge` 阶段 (Atlas Data Federation)阶段的一个版本，其中`connectionName`字段的值必须始终是连接注册表中远程集合的名称。
$vectorSearch	查询连接注册表中集合上的 MongoDB 向量搜索索引，并将结果附加到每个流媒体文档。

支持的聚合管道阶段

Atlas Stream Processing为核心MongoDB聚合管道语法提供了多个扩展。要学习；了解有关这些扩展的更多信息，请参阅支持的聚合管道阶段。

某些核心聚合管道阶段只有嵌套在其他管道阶段中时才受支持。下表显示了哪些聚合管道阶段支持在主管道内使用或嵌套在指定的聚合管道阶段内使用。

聚合阶段	在主管道中受支持	在 `$tumblingWindow`、`$hoppingWindow` 和 `$sessionWindow` 中受支持	在 `$https` 和 `$externalFunction` 中受支持
`$addFields`	√	√	√
`$project`	√	√	√
`$replaceRoot`	√	√	√
`$set`	√	√	√
`$redact`	√	√
`$match`	√	√
`$replaceWith`	√	√
`$unset`	√	√
`$unwind`	√	√
`$validate`	√	√
`$https`	√	√
`$externalFunction`	√	√
`$lookup` [1]	√	√
`$merge` [1]	√
`$tumblingWindow`	√
`$hoppingWindow`	√
`$sessionWindow`	√
$emit	√
$cachedLookup	√
$setStreamMeta	√
$throttle	√
`$group`		√
`$sort`		√
`$count`		√
`$limit`		√

[1]	( 1, 2 ) 支持使用修改后的语法。