當前位置：首頁 > 编程资源 > 编程问答 >内容正文

编程问答

使用storm 实时计算_使用Storm进行可扩展的实时状态更新

發布時間：2023/12/3 编程问答 59 豆豆

生活随笔收集整理的這篇文章主要介紹了使用storm 实时计算_使用Storm进行可扩展的实时状态更新小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

使用storm 實時計算

在本文中，我將說明如何借助Storm框架以可擴展且無鎖定的方式在數據庫中維護實時事件驅動流程的當前狀態。

Storm是基于事件的數據處理引擎。它的模型依賴于基本原語，例如事件轉換，過濾，聚合……，我們將它們組合成拓撲。拓撲的執行通常分布在多個節點上，并且風暴群集還可以并行執行給定拓撲的多個實例。因此，在設計時，必須牢記哪些Storm原語在分區范圍內執行，即在一個群集節點的級別上執行，以及哪些在群集范圍內執行（又稱為重新分區操作），因為它們涉及將事件從中移出的網絡流量。分區到分區）。 Storm Trident API文檔明確提到了哪些功能做什么，作用范圍如何。 Storm的分區概念與Kafka隊列的分區概念保持一致， Kafka隊列是入站事件的常見來源。

拓撲通常需要維護一些執行的持續狀態。例如，這可以是一些傳感器值的滑動窗口平均值，從推文中提取的近期情緒，在不同位置出現的人數。……由于某些狀態更新操作具有分區范圍（例如partitionAggregate ），因此可伸縮性模型在這里尤為重要。其他則具有集群范圍（例如groupby + perstitentAggregate的組合）。這篇文章中說明了這一點。

示例代碼在githup上可用。它基于Storm 0.8.2，Cassandra 1.2.5和JDK 1.7.0。請注意，該示例未包含適當的錯誤處理：噴口或螺栓均不支持重試失敗的元組，我將在以后的文章中解決。另外，我使用Java序列化將數據存儲在元組中，因此，即使Storm支持多種語言，我的示例也是特定于Java的。

實際示例：出席事件

我的示例是模擬一個跟蹤人們在建筑物內位置的系統。每當用戶進入或離開房間時，每個房間入口處的傳感器都會發出如下事件：

{"eventType": "ENTER", "userId": "John_5", "time": 1374922058918, "roomId": "Cafetaria", "id": "bf499c0bd09856e7e0f68271336103e0A", "corrId": "bf499c0bd09856e7e0f68271336103e0"} {"eventType": "ENTER", "userId": "Zoe_15", "time": 1374915978294, "roomId": "Conf1", "id": "3051649a933a5ca5aeff0d951aa44994A", "corrId": "3051649a933a5ca5aeff0d951aa44994"} {"eventType": "LEAVE", "userId": "Jenny_6", "time": 1374934783522, "roomId": "Conf1", "id": "6abb451d45061968d9ca01b984445ee8B", "corrId": "6abb451d45061968d9ca01b984445ee8"} {"eventType": "ENTER", "userId": "Zoe_12", "time": 1374921990623, "roomId": "Hall", "id": "86a691490fff3fd4d805dce39f832b31A", "corrId": "86a691490fff3fd4d805dce39f832b31"} {"eventType": "LEAVE", "userId": "Marie_11", "time": 1374927215277, "roomId": "Conf1", "id": "837e05916349b42bc4c5f65c0b2bca9dB", "corrId": "837e05916349b42bc4c5f65c0b2bca9d"} {"eventType": "ENTER", "userId": "Robert_8", "time": 1374911746598, "roomId": "Annex1", "id": "c461a50e236cb5b4d6b2f45d1de5cbb5A", "corrId": "c461a50e236cb5b4d6b2f45d1de5cbb5"}

對（“ ENTER”和“ LEAVE”）對中的每個事件與一個房間內一個用戶的一個占用時間段相對應。這可能對傳感器提出了很多要求，但是出于本示例的目的，這使我的生活更加輕松。

為了使事情變得有趣，讓我們想象一下，不能保證到達我們服務器的事件遵循時間順序（請參見生成事件的python腳本中的shuffle（）調用）。

我們將構建一個Storm拓撲，該拓撲將構建每個房間的每分鐘每分鐘的占用時間線，如本文結尾處的時間圖所示。在數據庫中，房間時間線被切成一個小時的時間段，這些時間段被獨立存儲和更新。這是Cafetaria占用1小時的示例：

{"roomId":"Cafetaria","sliceStartMillis":1374926400000,"occupancies":[11,12,12,12,13,15,15,14,17,18,18,19,20,22,22,22,21,22,23,25,25,25,28,28,33,32,31,31,29,28,27,27,25, 22,22,21,20,19,19,19,17,17,16,16,15,15,16,15,14,13,13,12,11,10,9,11,10,9,11,10]}

為了實現這一點，我們的拓撲需要：

根據correlationID重新組合“ ENTER”和“ LEAVE”事件，并為此用戶在此房間中產生相應的存在時間
將每個在場期間的影響應用于房間入住時間表

順便說一句，Cassandra提供了Counter列，盡管我可以很好地替代它們，但我在這里不使用它們。但是，我的目的是說明Storm功能，即使它會使方法有些虛構。

分組依據/ persistentAggregate / iBackingMap說明

在查看示例代碼之前，讓我們澄清一下這些“三叉戟風暴”原語如何協同工作。

想象一下，我們從上午9:47到上午10:34收到了兩個描述用戶在roomA中存在??的事件。更新會議室的時間表需要：

從數據庫加載兩個受影響的時間軸切片：[9.00am，10:00 am]和[10.00am，11:00 am]
在這兩個時間軸切片中添加此用戶的狀態
將它們保存到數據庫

但是，像這樣天真地實現此目標并不是最佳選擇，首先是因為它每個事件使用兩個DB請求，其次是因為這種“讀取-更新-寫入”序列通常需要一種鎖定機制，這種鎖定機制通常無法很好地擴展。

為了解決第一點，我們想為幾個事件重新組合數據庫操作。在Storm中，事件（或元組）被成批處理。 IBackingMap是一個我們可以實現的原語，它使我們可以立即查看整批元組。我們將使用它在批處理的開始（multiget）和結束時的所有DB-write操作（multiput）重新分組。但是，multiget不允許我們查看元組本身，而只能查看“查詢鍵”，這是根據元組內容計算出來的，如下所述。

原因在于上面提到的關于天真的實現的第二點：我們想并行執行幾個[multiget +更新邏輯+ multiput]流，而不依賴鎖。這是通過確保那些并行子流程更新不相交的數據集來實現的。這就要求定義拆分成并行流的拓撲元素還控制每個流內DB中要加載和更新的數據。該元素是Storm groupBy原語：它通過按字段值對元組進行分組來定義拆分，并且它通過將“ groupedBy”值作為對multiget的查詢關鍵字來控制每個并行流更新的數據。

下圖在房間占用示例中對此進行了說明（簡化為每個房間僅存儲一個時間線，而不是每個一小時的時間片一個時間線）：

但是，并行性并沒有完全發生（例如，當前的Storm實現在分組流中依次調用每個reducer / combiner），但這是設計拓撲時要牢記的一個好模型。

有趣的是，在groupBy和multiget之間發生了一些Storm魔術。回想一下，Storm旨在進行大規模分布，這意味著每個流在多個節點上并行執行，從諸如Hadoop HDFS或分布式Kafka隊列之類的分布式數據源獲取輸入數據。這意味著groupBy（）同時在多個節點上執行，所有可能處理的事件都需要組合在一起。 groupBy是一個重新分區操作，可確保將所有需要分組的事件發送到同一節點，并由IBackingMap +組合器或約簡器的同一實例處理，因此不會發生爭用情況。

同樣，Storm要求我們將IBackingMap包裝到可用的Storm MapState原語（或我們自己的原語）之一中，通常用于處理失敗/重播的元組。如上所述，我不在本文中討論這一方面。

使用這種方法，我們必須實現IBackingMap，以便它尊重以下屬性：

對于不同的鍵值，由multiget讀取和由IBackingMap的multiput操作寫入的數據庫行必須是不同的。

我想這就是他們將這些值稱為“關鍵”的原因（盡管任何尊重此屬性的方法都可以）。

回到例子

讓我們看看這在實踐中是如何工作的。該示例的主要拓撲在此處可用：

// reading events .newStream("occupancy", new SimpleFileStringSpout("data/events.json", "rawOccupancyEvent")) .each(new Fields("rawOccupancyEvent"), new EventBuilder(), new Fields("occupancyEvent"))

第一部分只是讀取JSON格式的輸入事件（我正在使用簡單的文件輸出），對它們進行反序列化，然后使用Java序列化將它們放入稱為“ occupancyEvent”的元組字段中。這些元組中的每一個都描述了用戶在房間內或房間外的“ ENTER”或“ LEAVE”事件。

// gathering "enter" and "leave" events into "presence periods" .each(new Fields("occupancyEvent"), new ExtractCorrelationId(), new Fields("correlationId")) .groupBy(new Fields("correlationId")) .persistentAggregate( PeriodBackingMap.FACTORY, new Fields("occupancyEvent"), new PeriodBuilder(), new Fields("presencePeriod")) .newValuesStream()

當我們遇到correlationId的不同值時，groupBy原語會創建盡可能多的元組組（這可能意味著很多，因為通常最多兩個事件具有相同的correlationId）。當前批處理中具有相同相關ID的所有元組將重新組合在一起，并且一組或幾組元組將一起呈現給persistentAggregate中定義的元素。 PeriodBackingMap是IBackingMap的實現，其中實現了multiget方法，該方法將接收下一步將要處理的元組組的所有相關ID（例如：{“ roomA”，“ roomB”，“ Hall ”}，如上圖所示）。

public List<RoomPresencePeriod> multiGet(List<List<Object>> keys) {return CassandraDB.DB.getPresencePeriods(toCorrelationIdList(keys)); }

該代碼只需要從數據庫中檢索每個相關ID的潛在存在期間即可。因為我們對一個元組字段進行了groupBy，所以每個List在這里都包含一個單個String：correlationId。請注意，我們返回的列表必須與鍵列表的大小完全相同，以便Storm知道哪個周期對應于哪個鍵。因此，對于數據庫中不存在的任何鍵，我們只需在結果列表中放置一個空值即可。

一旦加載，Storm就會將一個具有相同相關性ID的元組一個一個地呈現給我們的化簡器PeriodBuilder 。在我們的例子中，我們知道在此批次中，每個唯一的relativeId最多被調用兩次，但是一般來說可能更多，或者如果當前批次中不存在其他ENTER / LEAVE事件，則僅被調用一次。在對muliget（）/ multiput（）的調用與我們的reducer之間，借助我們選擇的MapState實現，Storm讓我們可以插入適當的邏輯來重放先前失敗的元組。在以后的文章中有更多的信息……

一旦我們減少了每個元組序列，Storm就會將結果傳遞給IBackingMap的mulitput（），在這里我們只是將所有內容“追加”到數據庫：

public void multiPut(List<List<Object>> keys, List<RoomPresencePeriod> newOrUpdatedPeriods) {CassandraDB.DB.upsertPeriods(newOrUpdatedPeriods); }

Storm persistenceAggregate使用我們的化簡提供給multitput（）的值，自動將其發送到拓撲元組的后續部分。這意味著我們剛剛建立的在線狀態很容易作為元組字段使用，我們可以使用它們直接更新會議室時間線：

// building room timeline .each(new Fields("presencePeriod"), new IsPeriodComplete()) .each(new Fields("presencePeriod"), new BuildHourlyUpdateInfo(), new Fields("roomId", "roundStartTime")) .groupBy(new Fields("roomId", "roundStartTime")) .persistentAggregate( TimelineBackingMap.FACTORY, new Fields("presencePeriod","roomId", "roundStartTime"), new TimelineUpdater(), new Fields("hourlyTimeline"))

第一行只是過濾掉尚未包含“ ENTER”和“ LEAVE”事件的任何期間。

然后， BuildHourlyUpdateInfo實現一對多的元組發射邏輯：對于每個占用期，它僅在“開始時間”內發射一個元組。例如，從9:47 am到10:34 am在roomA中的占用將在此處觸發針對RoomA的9.00am時間軸切片的元組的發射，以及另一個針對10.00am的元組的發射。

下一部分實現了與以前相同的groupBy / IBackingMap方法，只是這次使用了兩個分組鍵而不是一個（因此，mulitget中的List <Object>將包含兩個值：一個String和一個Long）。由于我們存儲一個小時的時間軸塊，因此上述IBackingMap的必要屬性得到了尊重。多重獲取為每個（“ roomId”，“開始時間”）對檢索時間線塊，然后TimelineUpdater （再次使用reducer）用與當前批次中找到的該時間線片相對應的每個存在時間更新時間線片（這就是BuildHourlyUpdateInfo的一對多元組發射邏輯）和multiput（）僅保存結果。

導致咖啡廳占用

當我們看著它時，一切總是更加美麗，所以讓我們來繪制房間的占用情況。稍加一些R代碼，我們就可以一分鐘一分鐘地看到房間的占用情況（這并不意味著什么，因為所有數據都是隨機的，但是……）：

結論

希望本文能為維護Storm拓撲中的狀態提供一種有用的方法。我還嘗試說明了將處理邏輯實現為小型拓撲元素的實現，將其彼此插入，而不是將一些“冗長的螺栓”捆綁在冗長而復雜的邏輯部分上。

Storm的一個重要方面是它的可擴展性，很可能去插入它的子類或在任何地方插入它的子類來調整其行為。春天有十年前的那種聰明而有趣的感覺（哦，該死，我現在有點老了……^ __ ^）

參考：來自Svend博客的 JCG合作伙伴 Svend Vanderveken 使用Storm進行的可伸縮實時狀態更新。

翻譯自: https://www.javacodegeeks.com/2013/08/scalable-real-time-state-update-with-storm.html