日前，MongoDB宣布推出Atlas Stream Processing公共預覽版。

在Atlas平臺上有興趣嘗試這項功能的開發者都享有完全的訪問權限，可前往“閱讀原文”鏈接點擊了解更多詳細信息或立即開始使用。

開發者喜歡文檔型數據庫的靈活性、易用性以及Query API查詢方式，能夠在MongoDB Atlas中以代碼方式處理數據。借助Atlas Stream Processing，MongoDB將這些相同的基本原則應用于流處理中。

Atlas Stream Processing于2023年美國紐約MongoDB用戶大會上首次推出，它旨在重塑聚合和豐富快速變化的事件數據流的體驗，并統一了處理流數據和靜態數據的方法。

在內測階段，MongoDB收到數千個開發團隊關于希望獲取訪問權限的請求，并且從數百個參與團隊中收集了有價值的反饋意見。其中一些用例包括：

● 某全球領先的航空公司利用復雜的聚合技術，快速處理維護和運營數據，以確保航班能夠準時起飛和到達，滿足成千上萬名乘客的需求；

● 某大型能源設備制造商使用Atlas Stream Processing來連續監控泵設備的海量數據，以避免意外停機并提升運行效果；

● 某創新型企業“軟件即服務”（SaaS）提供商充分利用Atlas Stream Processing中豐富的處理功能來及時提供包含背景信息的產品內警報，從而提升產品參與度。

這些用例僅僅是MongoDB在各行各業中觀察到的Atlas Stream Processing眾多應用實例中的一小部分。

除了MongoDB觀察到的眾多用例外，開發者也向MongoDB提供了豐富的見解，使MongoDB了解到他們希望未來Atlas Stream Processing應添加哪些功能。

除了通過變更流在Atlas數據庫中實現對數據的持續處理之外，許多開發者正在將Atlas Stream Processing與Kafka數據結合使用，這些Kafka數據被托管在Confluent Cloud、Amazon MSK、Azure Event Hubs和Redpanda等重要合作伙伴提供的服務中。在Atlas開發者數據平臺功能方面，MongoDB的目標始終是為開發者所依賴的關鍵技術提供更好的體驗。

隨著使用團隊數量的增加，MongoDB正在擴展功能，以便將在內測階段收集到的呼聲最高的反饋意見納入其中。

通過梳理大量的反饋意見，MongoDB從中總結出了三個共同的主題：提升開發者體驗、擴展高級特性和功能、改善運行和增強安全性。

一、提升開發者體驗

MongoDB在內測階段就將開發者體驗置于核心位置，這對于促使Atlas Stream Processing成為開發團隊的首選解決方案至關重要。在公共預覽版中，MongoDB更注重提升開發者的體驗，為此MongoDB增加了兩項增強型功能。

● VS Code集成。MongoDB VS Code插件增加了對連接流處理實例的支持。對于那些已經使用了該插件的開發者而言，隨著這項新功能的引入，團隊能夠在熟悉的開發環境中輕松地創建和管理處理器。這意味開發者不再需要頻繁切換工具，而可以將更多時間用來構建應用程序！

● 改進死信隊列（DLQ）功能。DLQ支持是實現強大流處理功能的關鍵要素，在公開預覽版中，MongoDB進一步擴展了DLQ功能。現在，當使用sp.process()來執行管道操作以及在運行中的處理器上運行.sample()時，DLQ消息將自動顯示，這樣可以簡化開發工作，而無需設置目標集合來充當DLQ。

二、擴展高級特性和功能

Atlas Stream Processing原本就已經支持很多常用的聚合操作符，這些操作符在靜態數據Query API查詢中經常被開發者所使用。而且MongoDB還增加了強大的窗口功能，以及可輕松合并數據并將其發送到Atlas數據庫或Kafka主題的功能。

公開預覽版將提供更多功能，以滿足那些依靠流處理來提供卓越客戶體驗的先進團隊的需求，包括：

●時間序列集合支持。Atlas Stream Processing現已支持將處理器的處理結果發送到時間序列集合中。這使得開發者能夠通過Atlas Stream Processing連續地對數據進行預處理，并且將數據保存下來以便將來在時間序列集合中進行回顧訪問（時間序列集合是Atlas平臺上專為高效存儲和查詢時間序列數據而構建的一種集合類型）。

● $lookup操作符。現在，開發者可以通過使用遠程Atlas集群的數據，對流處理器中正在處理的文檔以及目標集合中的字段進行連接，從而豐富流處理器中正在處理的文檔。

●變更流變更前后文檔信息的存儲。許多開發者正在使用Atlas Stream Processing通過變更流持續處理Atlas數據庫（作為源）中的數據。MongoDB在公共預覽中增強了變更流$source變量，以支持變更前和變更后文檔信息的存儲。這一功能使得開發者能夠處理一些常見的用例，包括計算文檔中字段之間的增量或差異，以及訪問已刪除文檔的完整內容等。

● 在合并和發出階段使用動態表達式進行條件路由。通過條件路由，開發者可以使用Atlas Stream Processing正在處理的文檔中的字段值，動態地將特定的消息發送到不同的Atlas集合或Kafka主題。現在$merge和$emit階段也支持使用動態表達式。基于此，用戶可以根據需求，在需要將信息分發到不同集合或主題的用例中使用Query API查詢。

● 空閑流超時。現在，用戶可以對那些因缺乏傳入數據導致水印無法更新的流進行配置，在特定時間段后關閉這些數據流，并輸出窗口結果。這對于處理數據流不一致的流媒體源來說至關重要。

三、改善運行和增強安全性

最后，在最近幾個月，MongoDB加大投入力度，改善Atlas Stream Processing的運行和安全。其中一些亮點包括：

● 檢查點。目前，Atlas Stream Processing通過執行檢查點的方式，在處理數據流過程中保存狀態信息。由于流處理器處于持續運行狀態，無論出現數據問題還是基礎設施故障，都需要一種智能的恢復機制來確保它能夠持續可靠地運行。通過采用檢查點機制，用戶可以輕松地從停止收集和處理數據的位置恢復流處理器的正常運行。

● Terraform提供商支持。Terraform目前支持創建連接和流處理實例（SPI）。這樣可以將基礎架構編寫為可重復部署的代碼。

● 安全角色。Atlas Stream Processing引入了項目級角色，為用戶提供了執行流處理任務所需的完整權限。流處理器能夠在特定角色的環境下運行，支持最低權限的配置選項。

● Kafka消費者群組支持。Atlas Stream Processing中的流處理器現在采用Kafka消費者群組來進行偏移跟蹤。得益于此，用戶可以輕松地調整處理器在流操作中的位置，并實時監控潛在的處理器延遲情況。

關于新功能的最后一點說明是，MongoDB計劃在Atlas Stream Processing的公開預覽版中，開始按照促銷價格收取費用，直至可廣泛使用的版本正式發布。更多關于Atlas Stream Processing價格的詳細信息，請參閱MongoDB的相關文檔。

對MongoDB而言，公共預覽版的推出是重要的一步，它標志著MongoDB的開發者數據平臺得到了擴展，為更多團隊提供了流處理解決方案。該解決方案旨在簡化構建反應式、響應式和事件驅動型應用程序的操作復雜性，同時提升開發者的體驗。