[TOC] # Storm簡介及安裝 ## 1 概述 > 在過去的十年里，數據處理發生了革命性的變化。MapReduce，Hadoop，以及相關的技術使我們可以存儲和處理以前不可想象規模的數據。很遺憾，這些數據處理系統都不是實時系統，命中注定也不是它們。根本沒辦法把Hadoop變成一個實時系統；實時數據處理和批處理的許多要求在根本上有很大不同。 > 然而，企業對大規模實時數據處理要求越來越多。缺乏“實時Hadoop”是數據處理生態系統中最大的窘境。 > Storm解決了這個窘境。 > Storm之前，你通常必須手動建立一個由許多隊列和許多worker組成的網絡來實現實時處理。worker處理隊列消息，更新數據庫，發送新消息給其它隊列以供后續處理。很遺憾， > 這種方法有很大的局限性： 1. 乏味：你大部份開發時間花費在配置消息發送，部署worker，部署中間隊列。你關心的實時處理邏輯對應到你的代碼的比例相對較小 。 2. 脆弱：沒有多少容錯。你負責保持每個worker和隊列正常工作。 3. 痛苦伸縮：當單個worker或隊列的消息吞吐量太高時，你需要分區，即數據如何分散。你需要重新配置其它worker，讓它們發送消息到新位置。這導致刪除或添加部件都可能失敗。 4. 雖然隊列+workers的范式能解決大量的消息，消息處理顯然是實時計算的基本范式。 > 問題是：你要怎么做，才能在某種程度上保證數據不會丟失，對海量消息輕松擴容，并且使用和運營工作都超級簡單呢？ > Storm滿足這些目標。 > Storm公開（expose）一組實時計算原語。類似MapReduce極大地簡化了編寫并行批處理程序，storm的原語極大地簡化了編寫并行實時計算程序。 > Storm的關鍵特性： 1. 用例非常廣泛：Storm可用于處理消息和更新數據庫（流處理），在數據流上進行持續查詢，并以流的形式返回結果到客戶端（持續計算），并行化一個類似實時查詢的熱點查詢（分布式的RPC），還有更多的用例。Storm的一組很小的原語滿足了驚人數量的用例。 2. 可伸縮：Storm隨時都可對大規模消息進行擴容。擴容一個拓撲，你只需要添加機器和增加的拓撲結構的并行設置。看一個storm規模的例子，一個storm集群有10個節點，一個最初的Storm應用每秒可以處理1,000,000個消息（指spout和bolt總共發射的消息總和），拓撲的其中一部分每秒數有數百個數據庫調用。Storm使用Zookeeper協調集群，使其集群可以擴容到非常大。 3. 保證數據不丟失：實時系統必須對成功處理數據提供有力保證 。系統丟棄數據的用例非常有限。Storm保證每個消息都被處理，這直接與其它系統截然不同，如S4。 4. 非常健壯：Storm與Hadoop不同，Hadoop難于管理早已臭名昭著，Storm集群只是干活。使用戶盡可能方便地管理storm集群是storm項目的一個明確目標。 5. 容錯：計算的執行過程中如果發生故障，Storm將在必要時重新分配任務。Storm確保計算永遠運行（或者直到你kill此計算） 。 6. 編程語言無關性：健壯和可伸縮的實時處理不應僅限于一個單一的平臺。Storm的拓撲結構和處理組件可以用任何語言定義，對任何人而言，Storm都是易接受的。 > 更多信息請參考：http://www.maoxiangyi.cn/index.php/archives/337 ## 2 Storm ### 1 特性 > 　　對比Hadoop的批處理，Storm是個實時的、分布式以及具備高容錯的計算系統。同Hadoop一樣Storm也可以處理大批量的數據，然而Storm在保證高可靠性的前提下還可以讓處理進行的更加實時;也就是說，所有的信息都會被處理。Storm同樣還具備容錯和分布計算這些特性，這就讓Storm可以擴展到不同的機器上進行大批量的數據處理。他同樣還有以下的這些特性： 1. 　　·易于擴展。對于擴展，你只需要添加機器和改變對應的topology(拓撲)設置。Storm使用Hadoop Zookeeper進行集群協調，這樣可以充分的保證大型集群的良好運行。 2. 　　·每條信息的處理都可以得到保證。 3. 　　·Storm集群管理簡易。 4. 　　·Storm的容錯機能：一旦topology遞交，Storm會一直運行它直到topology被廢除或者被關閉。而在執行中出現錯誤時，也會由Storm重新分配任務。 5. 　　·盡管通常使用Java，Storm中的topology可以用任何語言設計。 6. 　　當然為了更好的理解文章，你首先需要安裝和設置Storm。需要通過以下幾個簡單的步驟： 7. 　　·從Storm官方下載Storm安裝文件 8. 　　·將bin/directory解壓到你的PATH上，并保證bin/storm腳本是可執行的。 ### 2 Storm組件 > Storm集群主要由一個主節點和一群工作節點(worker node)組成，通過 Zookeeper進行協調。 > 　主節點： * 　　主節點通常運行一個后臺程序 —— Nimbus，用于響應分布在集群中的節點，分配任務和監測故障。這個很類似于Hadoop中的Job Tracker。 > 工作節點： * 　　工作節點同樣會運行一個后臺程序 —— Supervisor，用于收聽工作指派并基于要求運行工作進程。每個工作節點都是topology中一個子集的實現。而Nimbus和Supervisor之間的協調則通過Zookeeper系統或者集群。 ### 4 Zookeeper > Zookeeper是完成Supervisor和Nimbus之間協調的服務。而應用程序實現實時的邏輯則被封裝進Storm中的“topology”。topology則是一組由Spouts(數據源)和Bolts(數據操作)通過Stream Groupings進行連接的圖。 ### 4 編程模型 spout、Bolt、Stream Groupings > 　　Spout * 　　簡而言之，Spout從來源處讀取數據并放入topology。Spout分成可靠和不可靠兩種;當Storm接收失敗時，可靠的Spout會對tuple(元組，數據項組成的列表)進行重發;而不可靠的Spout不會考慮接收成功與否只發射一次。而Spout中最主要的方法就是nextTuple()，該方法會發射一個新的tuple到topology，如果沒有新tuple發射則會簡單的返回。 > 　　Bolt * 　　Topology中所有的處理都由Bolt完成。Bolt可以完成任何事，比如：連接的過濾、聚合、訪問文件/數據庫、等等。Bolt從Spout中接收數據并進行處理，如果遇到復雜流的處理也可能將tuple發送給另一個Bolt進行處理。而Bolt中最重要的方法是execute()，以新的tuple作為參數接收。不管是Spout還是Bolt，如果將tuple發射成多個流，這些流都可以通過declareStream()來聲明。 > 　　Stream Groupings * 　　Stream Grouping定義了一個流在Bolt任務間該如何被切分。這里有Storm提供的6個Stream Grouping類型： 1. 隨機分組(Shuffle grouping)：隨機分發tuple到Bolt的任務，保證每個任務獲得相等數量的tuple。 2. 字段分組(Fields grouping)：根據指定字段分割數據流，并分組。例如，根據“user-id”字段，相同“user-id”的元組總是分發到同一個任務，不同“user-id”的元組可能分發到不同的任務。 3. 全部分組(All grouping)：tuple被復制到bolt的所有任務。這種類型需要謹慎使用。 4. 全局分組(Global grouping)：全部流都分配到bolt的同一個任務。明確地說，是分配給ID最小的那個task。 5. 無分組(None grouping)：你不需要關心流是如何分組。目前，無分組等效于隨機分組。但最終，Storm將把無分組的Bolts放到Bolts或Spouts訂閱它們的同一線程去執行(如果可能)。 6. 直接分組(Direct grouping)：這是一個特別的分組類型。元組生產者決定tuple由哪個元組處理者任務接收。 > 　當然還可以實現CustomStreamGroupimg接口來定制自己需要的分組。 ## 3 Storm 偽分布式環境搭建 ### 1、下載安裝包 ~~~ wget http://124.202.164.6/files/1139000006794ECA/apache.fayea.com/storm/apache-storm-0.9.5/apache-storm-0.9.5.tar.gz ~~~ ### 2、解壓安裝包 ~~~ tar -zxvf apache-storm-0.9.5.tar.gz -C /export/servers/ cd /export/servers/ ln -s apache-storm-0.9.5 storm ~~~ ### 3、修改配置文件 ~~~ mv /export/servers/storm/conf/storm.yaml /export/servers/storm/conf/storm.yaml.bak vi /export/servers/storm/conf/storm.yaml ~~~ > 輸入以下內容：  ~~~ #指定storm使用的zk集群 storm.zookeeper.servers: - "zk01" - "zk02" - "zk03" #指定storm集群中的nimbus節點所在的服務器 nimbus.host: "storm01" #指定nimbus啟動JVM最大可用內存大小 nimbus.childopts: "-Xmx1024m" #指定supervisor啟動JVM最大可用內存大小 supervisor.childopts: "-Xmx1024m" #指定supervisor節點上，每個worker啟動JVM最大可用內存大小 worker.childopts: "-Xmx768m" #指定ui啟動JVM最大可用內存大小，ui服務一般與nimbus同在一個節點上。 ui.childopts: "-Xmx768m" #指定supervisor節點上，啟動worker時對應的端口號，每個端口對應槽，每個槽位對應一個worker supervisor.slots.ports: - 6700 - 6701 - 6702 - 6703 ~~~ > 注：配置文件每一行前面必須加一個‘空格’不然會有問題 ### 4、分發安裝包 ~~~ scp -r /export/servers/apache-storm-0.9.5 storm02:/export/servers ~~~ > 然后分別在各機器上創建軟連接 ~~~ cd /export/servers/ ln -s apache-storm-0.9.5 storm ~~~ ### 5、啟動集群 * 在nimbus.host所屬的機器上啟動 nimbus服務 ~~~ cd /export/servers/storm/bin/ nohup ./storm nimbus & ~~~ * 在nimbus.host所屬的機器上啟動ui服務 ~~~ cd /export/servers/storm/bin/ nohup ./storm ui & ~~~ * 在其它個點擊上啟動supervisor服務 ~~~ cd /export/servers/storm/bin/ nohup ./storm supervisor & ~~~ ### 6、查看集群 > 訪問nimbus.host:/8080，即可看到storm的ui界面。  ### 7、Storm常用操作命令 > 有許多簡單且有用的命令可以用來管理拓撲，它們可以提交、殺死、禁用、再平衡拓撲。 1. 提交任務命令格式：storm jar 【jar路徑】 【拓撲包名.拓撲類名】 【拓撲名稱】 ~~~ bin/storm jar examples/storm-starter/storm-starter-topologies-0.10.0.jar storm.starter.WordCountTopology wordcount ~~~ 2) 殺死任務命令格式：storm kill 【拓撲名稱】 -w 10（執行kill命令時可以通過-w [等待秒數]指定拓撲停用以后的等待時間） ~~~ storm kill topology-name -w 10 ~~~ 3) 停用任務命令格式：storm deactivte 【拓撲名稱】 ~~~ storm deactivte topology-name ~~~ > 我們能夠掛起或停用運行中的拓撲。當停用拓撲時，所有已分發的元組都會得到處理，但是spouts的nextTuple方法不會被調用。銷毀一個拓撲，可以使用kill命令。它會以一種安全的方式銷毀一個拓撲，首先停用拓撲，在等待拓撲消息的時間段內允許拓撲完成當前的數據流。 4) 啟用任務命令格式：storm activate【拓撲名稱】 ~~~ storm activate topology-name ~~~ 5) 重新部署任務命令格式：storm rebalance 【拓撲名稱】 ~~~ storm rebalance topology-name ~~~ > 再平衡使你重分配集群任務。這是個很強大的命令。比如，你向一個運行中的集群增加了節點。再平衡命令將會停用拓撲，然后在相應超時時間之后重分配worker，并重啟拓撲。 ## Storm編程快速入門資源圖片