Storm并發度怎么設置

本篇內容介紹了“Storm并發度怎么設置”的有關知識，在實際案例的操作過程中，不少人都會遇到這樣的困境，接下來就讓小編帶領大家學習一下如何處理這些情況吧！希望大家仔細閱讀，能夠學有所成！

　　Storm架構:master/slave

　　主節點:Nimbus

　　負責在集群上進行任務(Topology)的分發與資源的調度以及監控

　　工作節點:Supervisor

　　接收到任務請求后，啟動一個或多個Worker進程來處理任務;默認情況下，一個Supervisor最多啟動4個Worker

　　工作進程:Worker

　　在Supervisor中的子進程，存在著若干個Spout和Bolt線程，來負責Spout和Bolt組件處理任務(實際是開啟的executor線程)

　　作業:Topologies(死循環，不會結束)

　　Spout:獲取數據的組件

　　Bolt:處理數據的組件

　　Stream:Spout和Bolt之間數據流動的通道

　　Tuple:

　　1)Stream的最小組成單位，Spout向Bolt發送一次數據叫一個Tuple

　　2)同一個Stream中Tuple的類型相同，不同的Stream中可能相同/不同

　　3)一個key-value形式的Map

　　數據流分發策略(Stream groupings):

　　解決Spout和Bolt之間數據傳輸(發送Tuple元組)的問題

　　1)shuffleGrouping:

　　隨機派發Stream中的Tuple到Bolt中

　　2)fieldsGrouping:

　　根據字段的哈希值與Bolt個數進行取模操作然后進行分組發送，一個節點是一個Worker， 一個Bolt是一個task， 全部節點的Spout或Bolt的個數叫并發度。

　　Storm并發度設置:

　　1.Worker并發度:

　　首先按照集群規模和集群的物理位置來設定

　　一般會把Worker均分到每一個節點里， 一個supervisor默認設置一個Worker

　　2.Spout數量設定:

　　Spout總數默認等于Kafka(消息中間件)對應Topic的分區數，提高吞吐速度

　　一般一個Worker設置一個Spout

　　3.Bolt1數量設定:

　　首先根據數據量和處理數據的時間來設定

　　一般情況下， Bolt1的數量是Spout數量的2倍(根據項目進行修改)

　　4.Bolt2數量設定:

　　首先根據數據量和處理數據的時間來設定，因為Bolt1傳過來的中間結果數據已經減少很多，Bolt2的數量可以酌情減少。

　　容錯機制:異或方式<相同為，不同為1>

　　tupleId - 產生新數據，會產生一個tupleId;

　　整個過程中的tupleId按順序兩兩異或到最后

　　若結果為，則數據正確，否則錯誤

　　messageId - 代表整條信息，API中指定提供給程序員，long型

　　rootId - 代表某條信息，提供給storm框架

　　出現數據運算失敗的兩種情況:

　　execute(){

　　1.異常(數據異常)

　　2.任務運行超時 -- 認為處理失敗

　　}

　　因為數據發送時導致的數據重復發送問題， 如何解決?

　　Ⅰ.

　　1.比如對訂單信息做處理， 處理成功后， 把訂單信息ID存儲到Redis(set)

　　2.信息發送時， 判斷是否處理過此信息

　　execute(){

　　if()

　　else()

　　}

　　Ⅱ.

　　不作處理: 點擊流日日志分析: pv， uv

　　指標分析: 訂單人數， 訂單金額

　　消息的可靠性保障和acker機制: open / nextTuple / ack / fail/ close

　　Ⅰ.Spout類:

　　在發送tuple時，Spout會提供一個msgId，用于在后續識別tuple;Storm會根據msgId跟蹤創建的tuple樹，直到某個tuple被完整處理，根據msgId調用最初發送tuple的Spout中ack()方法，檢測到超時就調用fail()方法 -- 這兩個方法的調用必須由最初創建這個tuple的Spout執行;當Spout從消息隊列(Kafka/RocketMQ)中取出一條數據時，實際上沒有被取出，而是保持一個掛起狀態，等待消息完成的信號，掛起狀態的信息不會被發送到其它的消費者;當該消息被"取出"時，隊列會將消息體數據和一個唯一的msgId提供給客戶端，當Spout的ack()/fail()方法被調用時，Spout根據發送的id向隊列請求將消息從隊列中移除/重新放入隊列。

　　Ⅱ.acker任務:

　　高效的實現可靠性 -- 必須顯式的在Bolt中調用定義在Spout中的ack()和fail()方法，Storm拓撲有一些特殊的稱為"acker"的任務，負責跟蹤Spout發送的tuple的DAG，當一個acker發現DAG結束后，它就會給創建Spout tuple的Spout任務發送一條消息，讓這個任務來應答這個消息。acker并不會直接的跟蹤tuple樹，在acker樹中存儲了一個表，用于將Spout tuple的id與一對值相映射，id為創建這個tuple的任務id，第二個值為一個64bit的數字(ack val)，這個值是這棵樹中所有被創建的或者被應答的tuple的tuple id進行異或運算的結果值。

　　Ⅲ.移除可靠性:

　　1.將 Config.TOPOLOGY_ACKERS 設置為

　　2.在SpoutOutputCollector.emit 方法中省略消息 id 來關閉 spout tuple 的跟蹤功能

　　3.在發送 tuple 的時候選擇發送“非錨定”的(unanchored)tuple

“Storm并發度怎么設置”的內容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業相關的知識可以關注億速云網站，小編將為大家輸出更多高質量的實用文章！