Atlas Stream Processing

ATLAS

Atlas Stream Processing

简化 MongoDB 与 Apache Kafka 的集成，以便构建事件驱动型应用程序。

立即开始使用

为流媒体数据构建的数据模型

在处理流媒体数据时，模式管理对于确保数据的正确性以及提高开发者的工作效率至关重要。文档模型为开发者提供了一个灵活而直观的数据模型，便于构建包含实时数据的应用程序。

统一的开发者体验

开发者可以使用一个平台（跨 API、查询语言和数据模型）来持续处理来自 Apache Kafka 的流媒体数据，以及存储在数据库中的关键应用程序数据。

在 Atlas 中完全托管

仅需几行代码，开发者就能迅速将 Apache Kafka 的流媒体数据与数据库集成，构建出响应灵敏的应用程序——所有这些操作都在 Atlas 中完全托管。

MongoDB Atlas 中的原生流式传输

查看文档

使用 Atlas Stream Processing 轻松处理和验证复杂事件数据，将其准确合并到需要使用的地方。

查看文档

与 Apache Kafka 数据流进行集成

Atlas Stream Processing 使从 Apache Kafka 查询数据变得像查询 MongoDB 数据库一样简单。流处理器由源阶段、任意数量的处理阶段和汇阶段组成。

阅读文档

使用窗口功能进行连续分析

Atlas Stream Processing 中的窗口操作符允许您分析和处理连续数据流中固定大小的特定数据窗口，从而轻松地近乎实时地发现模式和趋势。

阅读文档

验证复杂事件的模式

在 Atlas Stream Processing 中，开发者可以进行持续验证。检测潜在的消息损坏和迟到数据，以确保事件在处理前正确形成。

阅读文档

与 Apache Kafka 数据流进行集成

Atlas Stream Processing 使从 Apache Kafka 查询数据变得像查询 MongoDB 数据库一样简单。流处理器由源阶段、任意数量的处理阶段和汇阶段组成。

阅读文档

MongoDB Query API

Code Snippet

阅读文档

使用窗口功能进行连续分析

Atlas Stream Processing 中的窗口操作符允许您分析和处理连续数据流中固定大小的特定数据窗口，从而轻松地近乎实时地发现模式和趋势。

阅读文档

MongoDB Query API

Code Snippet

阅读文档

验证复杂事件的模式

在 Atlas Stream Processing 中，开发者可以进行持续验证。检测潜在的消息损坏和迟到数据，以确保事件在处理前正确形成。

阅读文档

MongoDB Query API

Code Snippet

{ $validate: {
        验证器：{
            $and: [
                    {$expr: {
                        $ne: [
                        '$device_id'，
                        'device_8'
                        ]
                    }},
                    {$jsonSchema: {
                    required: [ 'device_id', 'timestamp', 'obs', 'event_type' ],
                    not : {required : ['event_details']},
                    properties: {
                        device_id: {
                            bsonType: 'string',
                            pattern: '^device_\\d+',
                            description: "'device_id' is required and must be like device_#"
                        },
                        obs: {
                            bsonType: 'object',
                            必需：[ 'watts', 'temp' ],
                            properties:{
                                瓦特：{
                                    bsonType : 'int',
                                    minimum: 0,
                                    最大值：250，
                                    description: "'obs.watts' is required and cannot be less then 0 or more then 250"
                                },
                                临时 : {
                                        bsonType: 'int',
                                        description: "'obs.temp' must be an integer"
                                },
                                }
                            },
                            event_type : {
                                    bsonType: 'int',
                                    minimum: 0,
                                    最大值：1,
                            },
                        时间戳：{
                            bsonType: 'string',
                            description: "'timestamp' must be a string "
                        }
                        }
                    }
                    }
            ]
                    }, validationAction : 'dlq'}
}

MongoDB Query API

Code Snippet

阅读文档

Atlas Stream Processing 客户成功案例

查看所有客户

持续见解

“Acoustic 的重点是通过行为见解为品牌赋能，使其能够创造卓越的个性化客户体验。通过 Atlas Stream Processing，我们的工程师就可以利用他们在 Atlas 中处理数据时已经掌握的技能来持续处理新数据，确保我们的客户能够获得实时的客户见解。”

John Riewerts
Acoustic 工程执行副总裁

了解详情

持续见解

John Riewerts
Acoustic 工程执行副总裁

了解详情

事件驱动的应用

“Atlas Stream Processing 让我们能够处理、验证和转换数据，再将数据发送到 AWS 中的消息传递架构，在整个平台上支持事件驱动的更新。Atlas Stream Processing 的可靠性和性能提高了我们的工作效率，改善了开发者体验，并降低了基础架构成本。”

Cody Perry
Meltwater 软件工程师

了解详情

学习资源中心

查找关于如何处理流媒体数据的白皮书、教程和视频。

流处理使用案例

查看所有使用案例

常见问答

各种来源不断生成流数据。物联网传感器、微服务和移动设备都是大量数据流的常见来源。流数据的连续性和不可变性使其与静态数据在数据库中的存储状态不同。

流处理不断从事件消息平台（如 Apache Kafka）摄取和转换事件数据，以执行各种功能。也就是说，创建简单的过滤器来删除不需要的数据，根据需要执行聚合来计数或汇总数据，创建有状态的窗口等等。在事件驱动型应用程序中，流处理可以成为差异化特性，从而实现反应更快、响应更迅速的客户体验。

了解详情

流数据存在于事件流平台（如 Apache Kafka）中，这些系统本质上是不可变的分布式日志。事件流平台使用 API 发布和消费事件数据。

开发者需要使用流处理器来执行更高级的处理，例如有状态的聚合、窗口操作、突变以及创建物化视图。这些操作与在数据库上运行查询时执行的操作类似，只不过流处理是对无穷无尽的数据流进行连续查询。流处理领域是一项新兴技术，Apache Flink 和 Spark Streaming 等解决方案正在迅速得到推广。

借助 Atlas Stream Processing，MongoDB 为开发者提供了一种更好的方法，利用聚合框架在应用程序中处理流。

了解详情

流处理是构建响应式、事件驱动型应用日益重要的组件。通过在 MongoDB Atlas 中添加流处理功能作为原生功能，我们将帮助更多开发者利用我们的多云开发者数据平台构建创新应用程序。

了解详情

流处理是持续发生的。在构建事件驱动型应用程序的背景下，流处理可实现卓越的反应式体验，如实时通知、个性化、路线规划和预测性维护。

批量处理不适用于持续生成的数据。相反，批量处理的处理方式是在指定时间内收集数据，然后根据需要处理这些静态数据。批量处理的一个示例是，零售企业在每天营业结束时收集销售数据，用于报告或更新库存水平。

了解详情

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

阅读文档

支持。当流处理器的最后一个操作完成后，Atlas Stream Processing 使用检查点文档来捕获流处理器的状态。这是在发生故障后重新启动处理器的关键功能，无需对数据进行大量的重新处理。

阅读文档

Atlas Stream Processing 支持 MongoDB Query API 中的 $validate 操作符。使用此操作符，开发者可以确保数据值和类型符合模式，并在数据不符合验证条件时选择将其路由到死信队列（DLQ）。