Atlas Stream Processing est désormais en aperçu public

Clark Gates-George and Joe Niemiec
February 13, 2024

Aujourd'hui, nous sommes ravis d'annoncer qu'Atlas Stream Processing est désormais disponible en aperçu public. Tous les développeurs qui travaillent avec Atlas peuvent ainsi l'essayer. Pour en savoir plus, consultez notre documentation ou lancez-vous dès aujourd'hui.

Écoutez le podcast MongoDB pour en savoir plus sur l'aperçu public d'Atlas Stream Processing avec Kenny Gorman, chef de produit Streaming.

Les développeurs aiment tout particulièrement la flexibilité et la facilité d'utilisation du modèle basé sur les documents, ainsi que la Requête API, qui leur permet de travailler avec des données en tant que code. Avec Atlas Stream Processing, nous appliquons ces mêmes principes fondamentaux au traitement des flux. Annoncé pour la première fois lors de la conférence .local NYC 2023, Atlas Stream Processing redéfinit l'expérience de l'agrégation et de l'enrichissement des flux de données d'événements à grande vitesse et à évolution rapide, et unifie les méthodes de travail des développeurs avec les données en mouvement et au repos.

Comment les développeurs utilisent-ils le produit jusqu'à présent ? Qu'avons-nous appris ?

Lors de l'aperçu privé, des milliers d'équipes de développement ont demandé à accéder à notre solution et nous avons recueilli des commentaires utiles de la part de centaines d'équipes engagées. Voici nos cas client :

Une compagnie aérienne internationale de premier plan utilise des agrégations complexes pour traiter rapidement les données relatives à la maintenance et aux opérations afin d'éviter les retards pour ses milliers de clients quotidiens.
Un grand fabricant d'équipements énergétiques qui utilise Atlas Stream Processing pour surveiller continuellement des données de pompe à volume élevé afin d'éviter les pannes et optimiser leurs rendements.
Un fournisseur SaaS innovant qui tire parti des capacités de traitement évoluées d’Atlas Stream Processing pour fournir des alertes contextuelles et dans les délais sur les produits afin d’améliorer l’engagement.

Ce ne sont que quelques exemples des nombreux cas d'utilisation observés dans différents secteurs d'activité. En outre, les développeurs nous donnent de nombreuses informations sur d'autres fonctionnalités d'Atlas Stream Processing qu'ils aimeraient utiliser.

Les développeurs ont non seulement pu bénéficier du traitement continu des données dans les bases de données Atlas via des change streams, mais ils ont aussi utilisé Atlas Stream Processing avec leurs données Kafka hébergées par des partenaires de renom tels que Confluent, Amazon MSK, Azure Event Hubs et Redpanda. Concernant les fonctionnalités de la developer data platform, nous nous sommes donné pour objectif d'améliorer continuellement l'expérience grâce aux technologies clés utilisées par les développeurs.

Quelles nouveautés offre l'aperçu public ?

Ce qui nous amène à évoquer les nouveautés. Plus nous adaptons nos solutions à d’autres équipes, plus nous étendons les fonctionnalités afin d'inclure celles qui sont le plus demandées dans notre aperçu privé. Parmi les nombreux commentaires reçus, trois problématiques sont ressorties :

Améliorer l'expérience des développeurs
Développer les fonctionnalités et fonctionnalités avancées
Améliorer les opérations et la sécurité

Améliorer l'expérience des développeurs

En avant-première privée, nous avons mis en place le cœur de l'expérience développeur, un enjeu clé pour faire d'Atlas Stream Processing une solution de choix pour les équipes de développement. Dans l'aperçu public, nous avons intégré deux améliorations supplémentaires :

Intégration du VS Code
Le plugin MongoDB VS Code intègre désormais la connexion aux instances de traitement des flux. Pour les développeurs qui utilisent déjà le plugin, cela permet aux équipes de créer et de gérer des processeurs dans un environnement de développement familier. Ainsi, vous perdez moins de temps à basculer entre les outils et consacrez plus de temps à la création de vos applications !
Amélioration des fonctionnalités de la Dead Letter Queue (DLQ)
La prise en charge de DLQ optimise le traitement des flux. Dans l'aperçu public, nous avons étendu ces fonctionnalités dans Atlas Stream Processing. Les messages Dead Letter Queue s'affichent désormais lors de l'exécution de pipelines avec sp.process () et lors de l'exécution de .sample () sur des processeurs en cours d'exécution. L'expérience de développement est ainsi plus rationalisée et ne nécessite pas la configuration d'une collection cible faisant office de DLQ.

Extension des fonctionnalités et fonctionnalités avancées

Atlas Stream Processing prenait déjà en charge la plupart des opérateurs d'agrégation clés maîtrisés par les développeurs dans la Requête API utilisée avec des données au repos. De puissantes fonctionnalités de fenêtrage et la possibilité de fusionner et d'émettre facilement des données vers la base de données Atlas ou vers un sujet Kafka ont également été ajoutées. L'aperçu public intégrera encore plus de fonctionnalités demandés par les équipes les plus expérimentées qui s'appuient sur le traitement des flux pour offrir des expériences client, à savoir :

$lookup
Les développeurs peuvent désormais enrichir les documents traités dans un processeur de flux avec des données provenant de clusters Atlas distants, en effectuant des liaisons entre les champs du document et de la collection cible.
Modifier les flux avant et après l'émission d'images
De nombreux développeurs utilisent Atlas Stream Processing pour traiter en continu les données des bases de données Atlas en tant que source via les change streams. Nous avons amélioré le change stream $source dans l'aperçu public avec la prise en charge des images avant et après. Les développeurs peuvent ainsi calculer les deltas entre les champs des documents et les cas d'utilisation nécessitant l'accès au contenu complet d'un document supprimé.
Routage conditionnel avec expressions dynamiques dans les étapes de fusion et d'émission
Le routage conditionnel permet aux développeurs d'utiliser la valeur des champs des documents traités dans Atlas Stream Processing pour envoyer dynamiquement des messages spécifiques à différentes collections Atlas ou sujets Kafka. Les étapes $merge et $emit prennent également désormais en charge l'utilisation d'expressions dynamiques. Il est ainsi possible d'utiliser l'API de requête pour des cas d'utilisation nécessitant la possibilité de faire basculer des messages vers différentes collections ou différents sujets, selon les besoins.
Délais d'inactivité des flux
Les flux sans filigrane avancé en raison d'un manque de données entrantes peuvent désormais être configurés pour se fermer automatiquement lorsque les résultats des fenêtres ont été émis depuis un certain temps. Cela peut s'avérer essentiel pour les sources de streaming dont les flux de données sont incohérents.

Améliorer les opérations et la sécurité

Enfin, nous avons beaucoup investi ces derniers mois en vue d'améliorer d'autres aspects opérationnels et sécuritaires d'Atlas Stream Processing. Voici quelques exemples :

Points de contrôle
Atlas Stream Processing effectue désormais des points de contrôle pour enregistrer l'état pendant le traitement. Les processeurs de flux exécutent continuellement des processus. Par conséquent, si un problème de données ou une défaillance de l'infrastructure surviennent, ils doivent disposer d'un mécanisme de récupération intelligent. Les points de contrôle facilitent la reprise des processeurs de flux à partir de l'endroit où les données ont cessé d'être collectées et traitées.
Prise en charge du fournisseur Terraform
La prise en charge de la création de connexions et d'instances de traitement de flux (SPI) est désormais disponible avec Terraform. Vous pouvez ainsi créer l’infrastructure sous forme de code pour reproduire les déploiements.
Rôles de sécurité
Atlas Stream Processing a ajouté un rôle au niveau du projet. Les utilisateurs disposent ainsi uniquement de l'autorisation nécessaire pour effectuer leurs tâches de traitement de flux. Les processeurs de flux peuvent s'exécuter dans le contexte d'un rôle spécifique, ce qui permet de configurer le modèle de moindre privilège.
Prise en charge des groupes de consommateurs Kafka
Les processeurs de flux dans Atlas Stream Processing utilisent désormais les groupes de consommateurs Kafka pour le suivi des décalages. Les utilisateurs peuvent ainsi changer facilement la position du processeur dans le flux pour les opérations et surveiller facilement les éventuels retards du processeur.

Dernière remarque concernant les nouveautés : des tarifs promotionnels vous seront offerts dans l'aperçu public d'Atlas Stream Processing en attendant le lancement en disponibilité générale. Pour en savoir plus sur la tarification d'Atlas Stream Processing, consultez notre documentation.

Créez votre premier processeur de stream dès aujourd'hui

L'aperçu public marque une avancée majeure pour nous, car nous développons la developer data platform et offrons à un plus grand nombre d'équipes une solution de traitement des flux qui simplifie la complexité opérationnelle de la création d'applications réactives, adaptées et événementielles, tout en offrant une expérience améliorée aux développeurs.

Nous avons hâte de voir vos créations !

Connectez-vous dès aujourd'hui pour commencer ou pour obtenir de plus amples informations sur Atlas Stream Processing dans notre documentation, nos ressources, nos tutoriels ou Learning Byte sur MongoDB University.

← Previous

Atlas Stream Processing 现已推出公共预览版

今天，我们很高兴地宣布，Atlas Stream Processing现已推出公共预览版。在Atlas平台上有兴趣尝试这项功能的开发者都享有完全的访问权限。参阅相关文档了解更多详细信息，或立即开始使用。欢迎收听MongoDB播客，聆听流媒体产品负责人Kenny Gorman对于Atlas Stream Processing公共预览版的详细介绍。开发者青睐文档模型的灵活性、易用性以及Query API查询方式，这使得他们能够在MongoDB Atlas中以代码方式处理数据。借助Atlas Stream Processing，我们将这些相同的基本原则应用于流处理中。Atlas Stream Processing在2023年纽约MongoDB用户大会上首次推出，它正在重塑旨在聚合和丰富快速变化的事件数据流的体验，并统一了处理流动数据和静态数据的方法。到目前为止，开发者使用该产品的情况如何？我们从中学到了什么？在内测阶段，我们收到了数千个开发团队关于希望获取访问权限的请求，并且从数百个参与团队中收集了有价值的反馈意见。其中一些用例包括：某全球领先的航空公司利用复杂的聚合技术，快速处理维护和运营数据，以确保航班能够准时起飞和到达，满足成千上万名乘客的需求；某大型能源设备制造商使用Atlas Stream Processing来连续监控关于泵设备的海量数据，以避免意外停机并提升运行效果；以及某创新型企业“软件即服务”（SaaS）提供商充分利用Atlas Stream Processing中丰富的处理功能来及时提供包含背景信息的产品内警报，从而提升产品参与度。这些用例仅仅是我们在各行各业中观察到的Atlas Stream Processing众多应用实例中的一小部分。除了我们观察到的众多用例外，开发者也向我们提供了丰富的见解，使我们了解到他们希望未来Atlas Stream Processing应添加哪些功能。除了支持通过 change stream 对 Atlas 数据库中的数据进行持续处理外，开发者可使用 Atlas Stream Processing 处理由 Confluent 、 Amazon MSK 、 Azure Event Hubs 和 Redpanda 等重要合作伙伴托管的 Kafka 数据，此功能也非常强大。我们开发者数据平台功能的目标始终是为开发者所依赖的关键技术提供更好的体验。公共预览版中有哪些新功能？基于上述情况，我们增加了新功能。随着使用团队数量的增加，我们正在扩展功能，以便将在内测阶段收集到的呼声最高的反馈意见纳入其中。通过梳理大量的反馈意见，我们从中总结出了三个共同的主题：提升开发者体验扩展高级特性和功能改善运行和增强安全性提升开发者体验在内测阶段，我们将开发者体验置于核心位置，这对于促使Atlas Stream Processing成为开发团队的首选解决方案至关重要。在公共预览版中，我们更注重提升开发者的体验，为此我们增加了两项增强型功能。 VS Code 集成 MongoDB VS Code插件增加了对连接流处理实例的支持。对于那些已经使用了该插件的开发者而言，随着这项新功能的引入，团队能够在熟悉的开发环境中轻松地创建和管理处理器。这意味开发者不再需要频繁切换工具，而可以将更多时间用来构建应用程序！改进了死信队列 (DLQ) 功能 DLQ 支持是实现强大流处理功能的关键要素，在公开预览版中，我们进一步扩展了DLQ功能。现在，当使用sp.process()来执行管道操作以及在运行中的处理器上运行.sample()时，DLQ消息将自动显示，这样可以简化开发工作，而无需设置目标集合来充当DLQ。扩展高级特性和功能 Atlas Stream Processing原本就已经支持很多常用的聚合操作符，这些操作符在静态数据Query API查询中经常被开发者所使用。而且我们还增加了强大的窗口功能，以及可轻松合并数据并将其发送到Atlas数据库或Kafka主题的功能。公开预览版将提供更多功能，以满足那些依靠流处理来提供卓越客户体验的先进团队的需求，包括： $lookup操作符现在，开发者可以通过使用远程Atlas集群的数据，对流处理器中正在处理的文档以及目标集合中的字段进行连接，从而丰富流处理器中正在处理的文档。变更流变更前后文档信息的存储许多开发者正在使用Atlas Stream Processing通过变更流持续处理Atlas数据库（作为源）中的数据。我们在公共预览中增强了变更流$source变量，以支持变更前和变更后文档信息的存储。这一功能使得开发者能够处理一些常见的用例，包括计算文档中字段之间的增量或差异，以及访问已删除文档的完整内容等。在合并和发出阶段使用动态表达式进行条件路由通过条件路由，开发者可以使用Atlas Stream Processing正在处理的文档中的字段值，动态地将特定的消息发送到不同的Atlas集合或Kafka主题。现在$merge和$emit阶段也支持使用动态表达式。基于此，用户可以根据需求，在需要将信息分发到不同集合或主题的用例中使用Query API查询。空闲流超时现在，用户可以对那些因缺乏传入数据导致水印无法更新的流进行配置，在特定时间段后关闭这些数据流，并输出窗口结果。这对于处理数据流不一致的流媒体源来说至关重要。改善运行和增强安全性最后，在最近几个月，我们加大投入力度，改善Atlas Stream Processing的运行和安全。其中一些亮点包括：检查点目前，Atlas Stream Processing通过执行检查点的方式，在处理数据流过程中保存状态信息。由于流处理器处于持续运行状态，无论出现数据问题还是基础设施故障，都需要一种智能的恢复机制来确保它能够持续可靠地运行。通过采用检查点机制，用户可以轻松地从停止收集和处理数据的位置恢复流处理器的正常运行。 Terraform提供商支持 Terraform目前支持创建连接和流处理实例（SPI）。这样可以将基础架构编写为可重复部署的代码。安全角色 Atlas Stream Processing引入了项目级角色，为用户提供了执行流处理任务所需的完整权限。流处理器能够在特定角色的环境下运行，支持最低权限的配置选项。 Kafka消费者群组支持 Atlas Stream Processing中的流处理器现在采用Kafka消费者群组来进行偏移跟踪。得益于此，用户可以轻松地调整处理器在流操作中的位置，并实时监控潜在的处理器延迟情况。关于新功能的最后一点说明是，我们计划在Atlas Stream Processing的公开预览版中，开始按照促销价格收取费用，直至可广泛使用的版本正式发布。更多关于Atlas Stream Processing价格的详细信息，请参阅我们的相关文档。即刻构建您的第一个流处理器对我们而言，公共预览版的推出是重要的一步，它标志着我们的开发者数据平台得到了扩展，为更多团队提供了流处理解决方案。该解决方案旨在简化构建反应式、响应式和事件驱动型应用程序的操作复杂性，同时提升开发者的体验。我们非常期待能够看到您构建的内容！立即登录以开始使用，或在我们的文档、资源、教程或 MongoDB University 上的 Learning Byte 中了解有关 Atlas Stream Processing 的更多信息。

February 13, 2024

Next →

AI-Powered Call Centers: A New Era of Customer Service

Customer satisfaction is critical for insurance companies. Studies have shown that companies with superior customer experiences consistently outperform their peers. In fact, McKinsey found that life and property/casualty insurers with superior customer experiences saw a significant 20% and 65% increase in Total Shareholder Return , respectively, over five years. A satisfied customer is a loyal customer. They are 80% more likely to renew their policies, directly contributing to sustainable growth. However, one major challenge faced by many insurance companies is the inefficiency of their call centers. Agents often struggle to quickly locate and deliver accurate information to customers, leading to frustration and dissatisfaction. This article explores how Dataworkz and MongoDB can transform call center operations. By converting call recordings into searchable vectors (numerical representations of data points in a multi-dimensional space), businesses can quickly access relevant information and improve customer service. We'll dig into how the integration of Amazon Transcribe, Cohere, and MongoDB Atlas Vector Search—as well as Dataworkz's RAG-as-a-service platform— is achieving this transformation. From call recordings to vectors: A data-driven approach Customer service interactions are goldmines of valuable insights. By analyzing call recordings, we can identify successful resolution strategies and uncover frequently asked questions. In turn, by making this information—which is often buried in audio files— accessible to agents, they can give customers faster and more accurate assistance. However, the vast volume and unstructured nature of these audio files make it challenging to extract actionable information efficiently. To address this challenge, we propose a pipeline that leverages AI and analytics to transform raw audio recordings into vectors as shown in Figure 1: Storage of raw audio files: Past call recordings are stored in their original audio format Processing of the audio files with AI and analytics services (such as Amazon Transcribe Call Analytics ): speech-to-text conversion, summarization of content, and vectorization Storage of vectors and metadata: The generated vectors and associated metadata (e.g., call timestamps, agent information) are stored in an operational data store Figure 1: Customer service call insight extraction and vectorization flow Once the data is stored in vector format within the operational data store, it becomes accessible for real-time applications. This data can be consumed directly through vector search or integrated into a retrieval-augmented generation (RAG) architecture, a technique that combines the capabilities of large language models (LLMs) with external knowledge sources to generate more accurate and informative outputs. Introducing Dataworkz: Simplifying RAG implementation Building RAG pipelines can be cumbersome and time-consuming for developers who must learn yet another stack of technologies. Especially in this initial phase, where companies want to experiment and move fast, it is essential to leverage tools that allow us to abstract complexity and don’t require deep knowledge of each component in order to experiment with and realize the benefits of RAG quickly. Dataworkz offers a powerful and composable RAG-as-a-service platform that streamlines the process of building RAG applications for enterprises. To operationalize RAG effectively, organizations need to master five key capabilities: ETL for LLMs: Dataworkz connects with diverse data sources and formats, transforming the data to make it ready for consumption by generative AI applications. Indexing: The platform breaks down data into smaller chunks and creates embeddings that capture semantics, storing them in a vector database. Retrieval: Dataworkz ensures the retrieval of accurate information in response to user queries, a critical part of the RAG process. Synthesis: The retrieved information is then used to build the context for a foundational model, generating responses grounded in reality. Monitoring: With many moving parts in the RAG system, Dataworkz provides robust monitoring capabilities essential for production use cases. Dataworkz's intuitive point-and-click interface (as seen in Video 1) simplifies RAG implementation, allowing enterprises to quickly operationalize AI applications. The platform offers flexibility and choice in data connectors, embedding models, vector stores, and language models. Additionally, tools like A/B testing ensure the quality and reliability of generated responses. This combination of ease of use, optionality, and quality assurance is a key tenet of Dataworkz's "RAG as a Service" offering. Diving deeper: System architecture and functionalities Now that we’ve looked at the components of the pre-processing pipeline, let’s explore the proposed real-time system architecture in detail. It comprises the following modules and functions (see Figure 2): Amazon Transcribe , which receives the audio coming from the customer’s phone and converts it into text. Cohere ’s embedding model, served through Amazon Bedrock , vectorizes the text coming from Transcribe. MongoDB Atlas Vector Search receives the query vector and returns a document that contains the most semantically similar FAQ in the database. Figure 2: System architecture and modules Here are a couple of FAQs we used for the demo: Q: “Can you explain the different types of coverage available for my home insurance?” A: “Home insurance typically includes coverage for the structure of your home, your personal belongings, liability protection, and additional living expenses in case you need to temporarily relocate. I can provide more detailed information on each type if you'd like.” Q: “What is the process for adding a new driver to my auto insurance policy?" A: “To add a new driver to your auto insurance policy, I'll need some details about the driver, such as their name, date of birth, and driver's license number. We can add them to your policy over the phone, or you can do it through our online portal.” Note that the question is reported just for reference, and it’s not used for retrieval. The actual question is provided by the user through the voice interface and then matched in real-time with the answers in the database using Vector Search. This information is finally presented to the customer service operator in text form (see Fig. 3). The proposed architecture is simple but very powerful, easy to implement, and effective. Moreover, it can serve as a foundation for more advanced use cases that require complex interactions, such as agentic workflows , and iterative and multi-step processes that combine LLMs and hybrid search to complete sophisticated tasks. Figure 3: App interface, displaying what has been asked by the customer (left) and how the information is presented to the customer service operator (right) This solution not only impacts human operator workflows but can also underpin chatbots and voicebots, enabling them to provide more relevant and contextual customer responses. Building a better future for customer service By seamlessly integrating analytical and operational data streams, insurance companies can significantly enhance both operational efficiency and customer satisfaction. Our system empowers businesses to optimize staffing, accelerate inquiry resolution, and deliver superior customer service through data-driven, real-time insights. To embark on your own customer service transformation, explore our GitHub repository and take advantage of the Dataworkz free tier .

November 27, 2024