概念 · Storm 1.1.0 中文文檔

# 概念本頁列出了Storm 的主要概念, 以及可以獲取到更多信息的資源鏈接, 概念如下： 1. Topologies（拓撲） 2. Streams（流） 3. Spouts 4. Bolts 5. Stream groupings（流分組） 6. Reliability（可靠性） 7. Tasks 8. Workers ### Topologies（拓撲）實時應用程序的邏輯被封裝在 Storm topology（拓撲）中. Storm topology（拓撲）類似于 MapReduce 作業. 兩者之間關鍵的區別是 MapReduce 作業最終會完成, 而 topology（拓撲）任務會永遠運行（除非 kill 掉它）. 一個拓撲是 Spout 和 Bolt 通過 stream groupings 連接起來的有向無環圖.這些概念會在下面的段落中具體描述. **相關資料:** * [TopologyBuilder](javadocs/org/apache/storm/topology/TopologyBuilder.html): Java中使用這個類來構建 topology（拓撲） * [如何在生產集群上運行 topologies（拓撲）](Running-topologies-on-a-production-cluster.html) * [如何使用 local 模式](Local-mode.html): 學習如何用本地模式開發和測試 topology（拓撲） ### Streams（流） stream 是 Storm 中的核心概念.一個 stream 是一個無界的、以分布式方式并行創建和處理的 Tuple 序列. stream 以一個 schema 來定義, 這個 schema 用來命名 stream tuple（元組）中的字段.默認情況下 Tuple 可以包含 integers, longs, shorts, bytes, strings, doubles, floats, booleans, and byte arrays 等數據類型.你也可以定義自己的 serializers, 以至于可以在 Tuple 中使用自定義的類型. 每一個流在聲明的時候會賦予一個 ID. 由于只包含一個 stream 的 Spout 和 Bolt 比較常見, [OutputFieldsDeclarer](javadocs/org/apache/storm/topology/OutputFieldsDeclarer.html) 有更方便的方法可以定義一個單一的 stream 而不用指定ID. 這個 stream 被賦予一個默認的 ID, "default". **相關資料:** * [Tuple](javadocs/org/apache/storm/tuple/Tuple.html): stream 由一系列連續的 Tuple 組成 * [OutputFieldsDeclarer](javadocs/org/apache/storm/topology/OutputFieldsDeclarer.html): 用于聲明 streams 和它的 schemas. * [Serialization](Serialization.html): Storm Tuple 的動態類型，和自定義 serializations 的相關信息 ### Spouts Spout 是一個 topology（拓撲）中 streams 的源頭. 通常 Spout 會從外部數據源讀取 Tuple，然后把他們發送到拓撲中（如 Kestel 隊列, 或者 Twitter API）. Spout 可以是 **可靠的** 或 **不可靠的**. 可靠的 Spout 在 Storm 處理失敗的時候能夠重放 Tuple, 不可靠的 Spout 一旦把一個 Tuple 發送出去就撒手不管了. Spout 可以發送多個流. 可以使用 [OutputFieldsDeclarer](javadocs/org/apache/storm/topology/OutputFieldsDeclarer.html) 的 declareStream 方法定義多個流, 在 [SpoutOutputCollector](javadocs/org/apache/storm/spout/SpoutOutputCollector.html) 對象的 emit 方法中指定要發送到的 stream . Spout 中的最主要的方法是 `nextTuple`. `nextTuple` 要么向 topology（拓撲）中發送一個新的 Tuple, 要么在沒有 Tuple 需要發送的情況下直接返回. 對于任何 Spout 實現, `nextTuple` 方法都必須非阻塞的, 因為 Storm 在一個線程中調用所有的 Spout 方法. Spout 的另外幾個重要的方法是 `ack` 和 `fail`. 這些方法在 Storm 檢測到 Spout 發送出去的 Tuple 被成功處理或者處理失敗的時候調用. `ack`和`fail`只會在可靠的 Spout 中調用. 更多相關信息, 請參見 [the Javadoc](javadocs/org/apache/storm/spout/ISpout.html). **相關資料:** * [IRichSpout](javadocs/org/apache/storm/topology/IRichSpout.html): 創建 Spout 時必須實現的接口 * [Guaranteeing message processing](Guaranteeing-message-processing.html) ### Bolts 拓撲中所有的業務處理都在 Bolts 中完成. Bolt 可以做很多事情，過濾, 函數, 聚合, 關聯, 與數據庫交互等. Bolt 可以做簡單 stream 轉換. 復雜的 stream 轉換一般需要多個步驟，因此也就要多個 Bolt 協同工作. 如, 轉換一個 tweets stream 為一個 trending images stream 需要兩個步驟：一個 Bolt 做每個圖片被收藏的滾動計數，同時一個或者多個 Bolt 輸出被收藏 Top X 的圖片 (你可以使用更具彈性的方式處理這個 stream 轉換, 用3個 Bolt 而不是先前的2個 Bolt ). Bolt 可以發送多個 stream. 可以使用 [OutputFieldsDeclarer](javadocs/org/apache/storm/topology/OutputFieldsDeclarer.html) 的 `declareStream` 方法定義多個 streams, 并且在使用 [OutputCollector](javadocs/org/apache/storm/task/OutputCollector.html) `emit` 方法的時候指定要發送的 stream. 當你聲明一個 Bolt 的 input stream，你總是會訂閱其他組件特定的 stream .如果你想要訂閱其他組件所有的 streams,你必須一個個的訂閱. [InputDeclarer](javadocs/org/apache/storm/topology/InputDeclarer.html) 有語法可以訂閱默認 stream-id 的 stream，代碼：`declarer.shuffleGrouping ("1")`，意思是：訂閱組件 “1” 的默認 stream, 等價于 `declarer.shuffleGrouping("1", DEFAULT_STREAM_ID)`. Bolt 中最主要的方法是 `execute` 方法, 當有一個新 Tuple 輸入的時候會進入這個方法. Bolt 使用[OutputCollector](javadocs/org/apache/storm/task/OutputCollector.html) 對象發送新的 Tuple. Bolt 必須在每一個 Tuple 處理完以后調用 `OutputCollector` 上的 `ack` 方法, Storm 就會知道 tuple 什么時候完成 (最終可以確定調用源 Spout Tuple 是沒有問題的). 當處理一個輸入的 Tuple：會基于這個 Tuple 產生零個或者多個 Tuple 發送出去，當所有的tuple 完成后，會調用 acking. Storm 提供了 [IBasicBolt](javadocs/org/apache/storm/topology/IBasicBolt.html) 接口會自動執行 acking . 最好在 Bolt 中啟動新的線程異步處理 tuples. [OutputCollector](javadocs/org/apache/storm/task/OutputCollector.html) 是線程安全的, 并且可以在任何時刻調用. **相關資料:** * [IRichBolt](javadocs/org/apache/storm/topology/IRichBolt.html): Bolts 的通用接口 * [IBasicBolt](javadocs/org/apache/storm/topology/IBasicBolt.html): 一個可以使用過濾或者一些簡單功能的 Bolt 的接口 * [OutputCollector](javadocs/org/apache/storm/task/OutputCollector.html): Bolts 使用這個類的實例發送 Tuple 到他們的輸出流 * [Guaranteeing message processing](Guaranteeing-message-processing.html) ### Stream groupings topology（拓撲）定義中有一部分是為每一個 bolt 指定輸入的 streams . stream grouping 定義了stream 如何在 Bolts tasks 之間分區. Storm 中一共有8個內置的 Stream Grouping. 可以通過實現 [CustomStreamGrouping](javadocs/org/apache/storm/grouping/CustomStreamGrouping.html) 接口來自定義 Stream groupings. 1. **Shuffle grouping**: Tuple 隨機的分發到 Bolt Task, 每個 Bolt 獲取到等量的 Tuple. 2. **Fields grouping**: streams 通過 grouping 指定的字段來分區. 例如流通過 "user-id" 字段分區, 具有相同 "user-id" 的 Tuple 會發送到同一個task, 不同 "user-id" 的 Tuple 可能會流入到不同的 tasks. 3. **Partial Key grouping**: stream 通過 grouping 中指定的 field 來分組, 與 Fields Grouping 相似. 但是對于 2 個下游的 Bolt 來說是負載均衡的, 可以在輸入數據不平均的情況下提供更好的優化. 以下地址 [This paper](https://melmeric.files.wordpress.com/2014/11/the-power-of-both-choices-practical-load-balancing-for-distributed-stream-processing-engines.pdf) 更好的解釋了它是如何工作的及它的優勢. 4. **All grouping**: stream 在所有的 Bolt Tasks之間復制. 這個 Grouping 小心使用. 5. **Global grouping**: 整個 stream 會進入 Bolt 其中一個任務.特別指出, 它會進入 id 最小的 task. 6. **None grouping**: 這個 grouping , 你不需要關心 stream 如何分組. 當前, None grouping 和 Shuffle grouping 等價. 同時, Storm 將使用 None grouping 的 bolts 和上游訂閱的 bolt和spout 運行在同一個線程 (when possible). 7. **Direct grouping**: 這是一種特殊的 grouping 方式. stream 用這個方式 group 意味著由這個 Tuple 的 **生產者** 來決定哪個 **消費者** 來接收它. Direct grouping 只能被用于 direct streams . 被發射到 direct stream 的 tuple 必須使用 [emitDirect](javadocs/org/apache/storm/task/OutputCollector.html#emitDirect)(int, int, java.util.List) 方法來發送. Bolt 可以使用 [TopologyContext](javadocs/org/apache/storm/task/TopologyContext.html) 或者通過保持對[OutputCollector](javadocs/org/apache/storm/task/OutputCollector.html)(返回 Tuple 被發送到的目標 task id) 中的 `emit` 方法輸出的跟蹤，獲取到它的所有消費者的 ID . 8. **Local or shuffle grouping**: 如果目標 Bolt 有多個 task 和 streams源在同一個 woker 進程中, Tuple 只會 shuffle 到相同 worker 的任務.否則, 就和 shuffle goruping 一樣. **相關資料:** * [TopologyBuilder](javadocs/org/apache/storm/topology/TopologyBuilder.html): 使用這個類來定義一個拓撲 * [InputDeclarer](javadocs/org/apache/storm/topology/InputDeclarer.html): 當在 `TopologyBuilder`上調用 `setBolt` 方法的時候返回這個對象, 用于聲明一個 Bolt 的 input streams 以及這些 streams 如何分組. ### 可靠性 Storm 保障每一個 Spout 的 Tuple 都會被 topology（拓撲）處理.通過跟蹤 tuples tree,每個 spout tuple 都會觸發 tree , 確保 tuples tree 成功完成. 每一個拓撲都有一個關聯的“message timeout”. 如果 Storm 檢測到一個 Spout Tuple 沒有在這個超時時間內被處理完成, 則判定這個 Tuple 失敗, 稍后重新執行. 要利用這個可靠性的功能, 當在 Tuple tree 中創建一個新的 edge ,必須告訴Storm,并且在一個單獨的 tuple 完成時也要通知 Storm. 以上操作在 Bolt 用于發送 Tuple 的 [OutputCollector](javadocs/org/apache/storm/task/OutputCollector.html) 對象中完成這個操作. Anchoring（錨點）在 `emit` 方法中完成, 使用 `ack` 方法來聲明你已經成功完成了一個 Tuple 的處理. 更詳細的說明, 請參閱 [保證消息容錯](Guaranteeing-message-processing.html). ### Tasks 每個 Spout 或者 Bolt 都以跨集群的多個 Task 方式執行. 每個 Task 對應一個 execution 的線程, stream groupings 定義如何從一個 Task 發送 Tuple 到另一個 Task. 可以在 [TopologyBuilder](http://storm.apache.org/releases/1.1.1/javadocs/org/apache/storm/topology/TopologyBuilder.html) 的`setSpout` 和 `setBolt` 方法中為每個 Spout 或者 Bolt 設置并行度,. ### Workers Topologies （拓撲）在一個或者跨多個 worker 執行. 每個 Worker 進程是一個物理的 JVM, 執行 topology（拓撲） Tasks 中的一個子集. 例如, 如果一個拓撲的并行度是 300, 共有 50 個 Worker 在運行, 每個 Worker 會分配到 6 個 Task（作為 Worker 中的線程）. Storm 會盡量把所有 Task 均勻的分配到所有的 Worker 上. **相關資料:** * [Config.TOPOLOGY_WORKERS](javadocs/org/apache/storm/Config.html#TOPOLOGY_WORKERS): 這個配置項設置用于運行 topology（拓撲）的 worker 數量.