【譯】Apache Storm系列之一（核心概念）

時間 2019-11-17

原文原文鏈接

本文列出 Storm 的幾個主要概念，並會給出相關資源的連接以便你獲取更多信息，概念主要以下：html

Topologies
Streams
Spouts
Bolts
Stream groupings
Reliability
Tasks
Workers

拓撲（Topologies）

實時應用程序的邏輯被打包到 Storm 拓撲中。一個 Storm 拓撲相似於一個 MapReduce 任務。關鍵的區別在於 MapReduce 任務最終會結束，而拓撲會一直運行（固然，除非你強制 kill 掉拓撲相關的進程）。拓撲能夠理解爲經過數據流（Stream Grouping）將 Spout 和 Bolt 相互鏈接而組成的圖狀結構的程序。spouts 和 bolts 的概念會在下文介紹。java

相關資源：數據庫

TopologyBuilder: 構造 topologies 的 Java 類
在生成環境的集羣上運行 topologies
本地模式: 學習如何在本地模式開發以及測試 topologies

流（Streams）

流是 Storm 的核心抽象。Storm中，一個流指的是在分佈式環境中被並行建立以及處理的元組（tuple）序列集。流是無限的元組（tuple）序列，以分佈式方式並行建立和處理。流每每有固定的模式（咱們稱之爲「fields」），不一樣模式由不一樣的元組（tuple）類型以必定的方式組成。一般，元組（tuple）能夠包含 integers, longs, shorts, bytes, strings, doubles, floats, booleans, 以及 byte arrays。固然，你也能夠經過定義可序列化的對象來實現自定義的元組類型。apache

相關資源：安全

Tuple: 流由元組組成
OutputFieldsDeclarer: 用來指定流以及它的模式
Serialization: 元組的動態類型以及自定義序列化

數據源（Spouts）

在拓撲中， spout 是流的來源。一般狀況，Spouts 會從外部源（例如消息隊列或者 Twitter API）讀取數據並將數據發送到拓撲中。Spouts 既能夠是可靠的，也能夠是不可靠的。可靠的狀況是若是數據流沒有被 Storm 處理，Spouts 將從新發送數據。不可靠的狀況則是對發送過的數據不予確認。微信

Spouts 一次能夠發送多個流。爲了實現多流發送，咱們可使用（實現） OutputFieldsDeclarer 接口中的 declareStream 方法來指定多個流，並使用（實現） SpoutOutputCollector 接口中的 emit 方法進行發送。異步

nextTuple 是 Spouts 中的主要方法。nextTuple 方法要麼發送一個新的元組到 topology 中，要麼直接返回（若是沒有新的元組須要發送）。須要注意的是，nextTuple 不該該被 Spout 的任何其餘方法所阻塞，不然會致使數據流的中止接入，這是由於 Spout 的全部方法是在一個線程中執行。分佈式

ack 和 fail 是 Spouts 中另外兩個重要的方法。Spouts 爲可靠模式時，Storm 會檢測每個從 Spouts 發送出去的元組是否成功，成功調用 ack，失敗調用 fail。固然，在不可靠模式下，是不會調用這兩個方法的。學習

相關資源：測試

IRichSpout: 自定義的 spouts 必須實現這個接口
消息的可靠性處理

處理組件（Bolts）

topologies 全部的處理都是在 bolts 中進行。bolts 能夠作不少事情，例如：過濾流、邏輯處理、聚合、鏈接、數據庫交互等等。

bolts 能夠從事簡單的數據流轉換。處理複雜的數據流轉換一般須要將流程分紅多步，這也就意味着咱們可使用多類（個） bolt。例如，從微博數據流中得出一個趨勢圖，實現這個需求咱們至少須要兩步：第一個 bolt 計算每一個圖片的點擊數，第二個 bolt 在第一個基礎上得出 TOP X 的圖片（固然爲了流程可擴展，咱們可使用更多的 bolt,不只限於兩個）。

bolts 一次能夠發送多個流。爲了實現多流發送，咱們可使用（實現） OutputFieldsDeclarer 接口中的 declareStream 方法來指定多個流，並使用（實現） OutputCollector 接口中的 emit 方法進行發送（跟 spout 相似）。

在定義 Bolt 的輸入數據流時，你須要從其餘的 Storm 組件中訂閱指定的數據流。若是你須要從其餘全部的組件中訂閱數據流，你就必需要在定義 Bolt 時分別註冊每個組件。對於聲明爲默認 id 的數據流，InputDeclarer 接口有訂閱此類數據流的語法糖。調用 declarer.shuffleGrouping("1") 將訂閱來自 id 爲「1」的組件（spout/bolt）產生的數據流，其等價於調用 declarer.shuffleGrouping("1", DEFAULT_STREAM_ID)

execute 是 bolt 的主要方法，它接收新的元組做爲輸入。bolt 使用 OutputCollector 對象來發送新的元組。bolt 必須爲每一個經由它處理的元組調用 OutputCollector 中的 ack 方法，這樣以便 Storm 知道這些元組何時被處理完成（最終判斷對原始 spout 元組的響應是否合適）。處理元組的通常狀況是，咱們能夠發送多個元組或者直接不發送，而後響應下一個輸入元組，咱們能夠實現 IBasicBolt 接口來完成 bolt 操做。

咱們能夠在 bolt 任務中開啓一個新的線程來完成異步操做。OutputCollector 線程安全而且能夠隨時被調用。

相關資源：

IRichBolt: bolt 的主要接口.
IBasicBolt: 另一個簡單接口，幫助用實現過濾以及其餘功能
OutputCollector: 使用這個類來發送元組到輸出流中
消息的可靠性處理

流分組（Stream groupings）

定義一個 topology 的重要一部分是指定每一個 bolt 應該接收哪些流做爲輸入。流分組（stream grouping）定義了流如何分發到各個 bolt 中。

Storm 提供了 8 種流分組策略。固然，你也能夠經過實現 CustomStreamGrouping 接口來實現一個用戶自定義的流分組：

Shuffle grouping : 元組被隨機分發到各個 bolt 任務中，也就是說每一個 bolt 接收到大體相同數目的元組。
Fields grouping : 根據指定的 field 進行分組，同一個 field 的值必定會被髮送到同一個 task 上。例如，若是流按照 "user-id" 這個 field 進行分組，那麼相同的 "user-id" 值會進入相同的任務（task），若是不一樣，則進入不一樣的任務。
Partial Key grouping : 與 Fields grouping 相似，根據指定的 field 的一部分進行分組分發，可以很好地實現 load balance，將元組發送給下游的 bolt 對應的 task，特別是在存在數據傾斜的場景，使用 Partial Key grouping 可以更好地提升資源利用率
All grouping : 流複製到全部 bolt task 上。
Global grouping: 全部的流都指向一個 bolt 的同一個 task，也就是Task ID最小的。
None grouping : 使用這個分組，用戶不用關心流是如何進行分組的。目前，這個分組相似於 Shuffle grouping。不過將來 Storm 可能會考慮經過這種分組來讓 Bolt 和它所訂閱的 Spout 或 Bolt 在同一個線程中執行。
Direct grouping : 由 tupe 的生產者來決定發送給下游的哪個 bolt 的 task ，這個要在實際開發編寫 bolt 代碼的邏輯中進行精確控制。
Local or shuffle grouping : 若是目標 bolt 有1個或多個 task 都在同一個 worker 進程對應的 JVM 實例中，則 tuple 只發送給這些 task。

可靠性（Reliability）

Storm 保證每一個 spout 元組都能在拓撲中被處理。經過跟蹤由 Spout 發出的每一個元組構成的元組樹能夠肯定元組是否已經完成處理。每一個拓撲都有與之相關的消息超時。若是在超時時間內沒有檢測到元組是否被完整處理，該原則將會被標記並從新發送。

想要使用 Storm 這個可靠性功能，你必須在元組建立以及處理完成時告訴 Storm。你可使用用於發送數據流的 OutputCollector 對象，並使用 ack 方法代表你已經完成了元組的處理。

任務（Tasks）

集羣中，每個 spout 和 bolt 運行了多個任務。每一個任務對應一個執行線程，流分組定義如何將元組從一組任務發送到另外一組任務。你可使用 TopologyBuilder 中的 setSpout 和 setBolt 方法來設置任務並行度。

Workers

一個拓撲中運行了一個或多個 worker 進程。每一個進程都是一個物理 JVM，而且拓撲中的全部 task 都在這些進程中執行。例如，若是並行度爲 300，咱們有 50 個worker 進程，那麼每一個進程將處理 6 個 task。Storm 有其機制致力於將全部任務儘可能平均地分配到每一個進程中。

相關資源： Config.TOPOLOGY_WORKERS: 設置 worker 數量的配置