Atlas Stream Processing使您能够使用Atlas数据库中使用的相同聚合操作来读取、写入和转换数据流。Atlas Stream Processing允许您：
构建 聚合管道以持续处理流媒体数据。
执行持续验证，以检查消息格式是否正确、检测消息损坏并检测延迟到达的数据。
在文档流经管道时转换字段，并使用每个文档中的字段或表达式作为键将这些文档路由到不同的数据库、 Kafka主题或其他外部接收器。
不断将结果发布到Atlas集合或Apache Kafka 集群，确保最新的数据视图和分析。
Atlas Stream Processing 组件直接属于 Atlas 项目，并独立于 Atlas 集群运行。
流媒体数据
流是源自一个或多个源的不可变数据的连续流。数据流的示例包括传感器的温度或压力读数、金融交易记录或变更数据捕获事件。
数据流源自Apache Kafka主题或MongoDB变更流等来源。然后，您可以将处理后的数据写入接收器，包括Apache Kafka主题、Atlas集合、外部函数或云数据存储。
Atlas Stream Processing提供原生流处理功能，可对连续数据进行操作，而不受静态数据库的时间和计算限制。
流处理器的结构
流处理器采用管道的形式，从概念上可以分为三个阶段。
来源
流处理器首先从Atlas Stream Processing连接的流媒体数据源摄取文档。这些可以是像Apache Kafka这样的代理系统，也可以是像Atlas读/写入操作生成的数据库变更流这样的数据库变更流。这些输入必须是有效的 json 或 ejson 文档。一旦$source阶段摄取文档，您就可以对该文档应用MongoDB聚合，以根据需要对其进行转换。
除了从流媒体源摄取数据外， Atlas Stream Processing还支持使用来自HTTPS 请求的数据和 $lookup 操作来丰富文档，以连接来自已连接Atlas集群的数据。
管道
除了标准的MongoDB聚合操作符和阶段套件之外，流处理器还利用聚合管道阶段和聚合操作符来转换提取的数据并提取有价值的见解。Atlas Stream Processing可以将其无法进程的文档写入死信队列。
您可以通过重组文档、添加或删除字段、从集合中查找信息等来丰富文档。Atlas Stream Processing还支持您使用Windows收集事件并执行任意函数。
Windows
Windows是在设立时间段内聚合流媒体数据的管道阶段。这使您能够对数据群组、取平均值、查找最小值和最大值以及执行各种其他操作，否则这些操作不适用于流媒体数据。每个流处理器只能有一个窗口阶段。
功能
Atlas Stream Processing支持调用自定义JavaScript 函数 或 Amazon Web Services Lambda 函数，这些函数会针对流处理传递给它们的每个文档运行。
接收器
处理完摄取的数据后，流处理器会持久将其写入接收器。Atlas Stream Processing提供 $emit 和 $merge 阶段，用于写入不同的接收器类型。这些阶段是互斥的，并且每个流处理器只能有一个 Sink 阶段。您的管道可以包含将处理后的文档写入同一接收器连接中的不同Kafka主题或Atlas集合的逻辑。
Atlas Stream Processing 区域
Atlas Stream Processing支持在Amazon Web Services、Azure和Google Cloud Platform上创建流处理实例。有关可用区域的列表，请参阅以下内容的流处理实例部分：
流处理器可以读取和写入托管在不同云提供商或不同区域中的集群。
帐单
有关计费的信息，请参阅Atlas Stream Processing计费页面。
后续步骤
要开始实际操作Atlas Stream Processing，请参阅Atlas Stream Processing入门。
有关Atlas Stream Processing核心概念的更多详细信息，请参阅以下内容：