Storm 系列（五）—— Storm 程式設計模型詳解
一、簡介
下圖為 Strom 的執行流程圖，在開發 Storm 流處理程式時，我們需要採用內建或自定義實現 spout(資料來源) 和 bolt(處理單元)，並通過 TopologyBuilder 將它們之間進行關聯，形成 Topology。
二、IComponent介面
IComponent 介面定義了 Topology 中所有元件 (spout/bolt) 的公共方法，自定義的 spout 或 bolt 必須直接或間接實現這個介面。
public interface IComponent extends Serializable { /** * 宣告此拓撲的所有流的輸出模式。 * @param declarer 這用於宣告輸出流 id，輸出欄位以及每個輸出流是否是直接流（direct stream） */ void declareOutputFields(OutputFieldsDeclarer declarer); /** * 宣告此元件的配置。 * */ Map<String, Object> getComponentConfiguration(); }
三、Spout
3.1 ISpout介面
自定義的 spout 需要實現 ISpout 介面，它定義了 spout 的所有可用方法：
public interface ISpout extends Serializable { /** * 元件初始化時候被呼叫 * * @param conf ISpout 的配置 * @param context 應用上下文，可以通過其獲取任務 ID 和元件 ID，輸入和輸出資訊等。 * @param collector 用來發送 spout 中的 tuples，它是執行緒安全的，建議儲存為此 spout 物件的例項變數 */ void open(Map conf, TopologyContext context, SpoutOutputCollector collector); /** * ISpout 將要被關閉的時候呼叫。但是其不一定會被執行，如果在叢集環境中通過 kill -9 殺死程序時其就無法被執行。 */ void close(); /** * 當 ISpout 從停用狀態啟用時被呼叫 */ void activate(); /** * 當 ISpout 停用時候被呼叫 */ void deactivate(); /** * 這是一個核心方法，主要通過在此方法中呼叫 collector 將 tuples 傳送給下一個接收器，這個方法必須是非阻塞的。 * nextTuple/ack/fail/是在同一個執行緒中執行的，所以不用考慮執行緒安全方面。當沒有 tuples 發出時應該讓 * nextTuple 休眠 (sleep) 一下，以免浪費 CPU。 */ void nextTuple(); /** * 通過 msgId 進行 tuples 處理成功的確認，被確認後的 tuples 不會再次被髮送 */ void ack(Object msgId); /** * 通過 msgId 進行 tuples 處理失敗的確認，被確認後的 tuples 會再次被髮送進行處理 */ void fail(Object msgId); }
3.2 BaseRichSpout抽象類
通常情況下，我們實現自定義的 Spout 時不會直接去實現 ISpout 介面，而是繼承 BaseRichSpout。BaseRichSpout 繼承自 BaseCompont，同時實現了 IRichSpout 介面。
IRichSpout 介面繼承自 ISpout 和 IComponent,自身並沒有定義任何方法：
public interface IRichSpout extends ISpout, IComponent {
}BaseComponent 抽象類空實現了 IComponent 中 getComponentConfiguration
public abstract class BaseComponent implements IComponent {
@Override
public Map<String, Object> getComponentConfiguration() {
return null;
}
}BaseRichSpout 繼承自 BaseCompont 類並實現了 IRichSpout 介面，並且空實現了其中部分方法：
public abstract class BaseRichSpout extends BaseComponent implements IRichSpout {
@Override
public void close() {}
@Override
public void activate() {}
@Override
public void deactivate() {}
@Override
public void ack(Object msgId) {}
@Override
public void fail(Object msgId) {}
}通過這樣的設計，我們在繼承 BaseRichSpout 實現自定義 spout 時，就只有三個方法必須實現：
- open ： 來源於 ISpout，可以通過此方法獲取用來發送 tuples 的
SpoutOutputCollector； - nextTuple ：來源於 ISpout，必須在此方法內部發送 tuples；
- declareOutputFields ：來源於 IComponent，聲明發送的 tuples 的名稱，這樣下一個元件才能知道如何接受。
四、Bolt
bolt 介面的設計與 spout 的類似：
4.1 IBolt 介面
/**
* 在客戶端計算機上建立的 IBolt 物件。會被被序列化到 topology 中（使用 Java 序列化）,並提交給叢集的主機（Nimbus）。
* Nimbus 啟動 workers 反序列化物件，呼叫 prepare，然後開始處理 tuples。
*/
public interface IBolt extends Serializable {
/**
* 元件初始化時候被呼叫
*
* @param conf storm 中定義的此 bolt 的配置
* @param context 應用上下文，可以通過其獲取任務 ID 和元件 ID，輸入和輸出資訊等。
* @param collector 用來發送 spout 中的 tuples，它是執行緒安全的，建議儲存為此 spout 物件的例項變數
*/
void prepare(Map stormConf, TopologyContext context, OutputCollector collector);
/**
* 處理單個 tuple 輸入。
*
* @param Tuple 物件包含關於它的元資料（如來自哪個元件/流/任務）
*/
void execute(Tuple input);
/**
* IBolt 將要被關閉的時候呼叫。但是其不一定會被執行，如果在叢集環境中通過 kill -9 殺死程序時其就無法被執行。
*/
void cleanup();4.2 BaseRichBolt抽象類
同樣的，在實現自定義 bolt 時，通常是繼承 BaseRichBolt 抽象類來實現。BaseRichBolt 繼承自 BaseComponent 抽象類並實現了 IRichBolt 介面。
IRichBolt 介面繼承自 IBolt 和 IComponent,自身並沒有定義任何方法：
public interface IRichBolt extends IBolt, IComponent {
}通過這樣的設計，在繼承 BaseRichBolt 實現自定義 bolt 時，就只需要實現三個必須的方法：
- prepare： 來源於 IBolt，可以通過此方法獲取用來發送 tuples 的
OutputCollector； - execute：來源於 IBolt，處理 tuples 和傳送處理完成的 tuples；
- declareOutputFields ：來源於 IComponent，聲明發送的 tuples 的名稱，這樣下一個元件才能知道如何接收。
五、詞頻統計案例
5.1 案例簡介
這裡我們使用自定義的 DataSourceSpout 產生詞頻資料，然後使用自定義的 SplitBolt 和 CountBolt 來進行詞頻統計。
案例原始碼下載地址：storm-word-count
5.2 程式碼實現
1. 專案依賴
<dependency>
<groupId>org.apache.storm</groupId>
<artifactId>storm-core</artifactId>
<version>1.2.2</version>
</dependency>2. DataSourceSpout
public class DataSourceSpout extends BaseRichSpout {
private List<String> list = Arrays.asList("Spark", "Hadoop", "HBase", "Storm", "Flink", "Hive");
private SpoutOutputCollector spoutOutputCollector;
@Override
public void open(Map map, TopologyContext topologyContext, SpoutOutputCollector spoutOutputCollector) {
this.spoutOutputCollector = spoutOutputCollector;
}
@Override
public void nextTuple() {
// 模擬產生資料
String lineData = productData();
spoutOutputCollector.emit(new Values(lineData));
Utils.sleep(1000);
}
@Override
public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
outputFieldsDeclarer.declare(new Fields("line"));
}
/**
* 模擬資料
*/
private String productData() {
Collections.shuffle(list);
Random random = new Random();
int endIndex = random.nextInt(list.size()) % (list.size()) + 1;
return StringUtils.join(list.toArray(), "\t", 0, endIndex);
}
}上面類使用 productData 方法來產生模擬資料，產生資料的格式如下：
Spark HBase
Hive Flink Storm Hadoop HBase Spark
Flink
HBase Storm
HBase Hadoop Hive Flink
HBase Flink Hive Storm
Hive Flink Hadoop
HBase Hive
Hadoop Spark HBase Storm3. SplitBolt
public class SplitBolt extends BaseRichBolt {
private OutputCollector collector;
@Override
public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
this.collector=collector;
}
@Override
public void execute(Tuple input) {
String line = input.getStringByField("line");
String[] words = line.split("\t");
for (String word : words) {
collector.emit(new Values(word));
}
}
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("word"));
}
}4. CountBolt
public class CountBolt extends BaseRichBolt {
private Map<String, Integer> counts = new HashMap<>();
@Override
public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
}
@Override
public void execute(Tuple input) {
String word = input.getStringByField("word");
Integer count = counts.get(word);
if (count == null) {
count = 0;
}
count++;
counts.put(word, count);
// 輸出
System.out.print("當前實時統計結果:");
counts.forEach((key, value) -> System.out.print(key + ":" + value + "; "));
System.out.println();
}
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
}
}5. LocalWordCountApp
通過 TopologyBuilder 將上面定義好的元件進行串聯形成 Topology，並提交到本地叢集（LocalCluster）執行。通常在開發中，可先用本地模式進行測試，測試完成後再提交到伺服器叢集執行。
public class LocalWordCountApp {
public static void main(String[] args) {
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("DataSourceSpout", new DataSourceSpout());
// 指明將 DataSourceSpout 的資料傳送到 SplitBolt 中處理
builder.setBolt("SplitBolt", new SplitBolt()).shuffleGrouping("DataSourceSpout");
// 指明將 SplitBolt 的資料傳送到 CountBolt 中 處理
builder.setBolt("CountBolt", new CountBolt()).shuffleGrouping("SplitBolt");
// 建立本地叢集用於測試 這種模式不需要本機安裝 storm,直接執行該 Main 方法即可
LocalCluster cluster = new LocalCluster();
cluster.submitTopology("LocalWordCountApp",
new Config(), builder.createTopology());
}
}6. 執行結果
啟動 WordCountApp 的 main 方法即可執行，採用本地模式 Storm 會自動在本地搭建一個叢集，所以啟動的過程會稍慢一點，啟動成功後即可看到輸出日誌。
六、提交到伺服器叢集執行
6.1 程式碼更改
提交到伺服器的程式碼和原生代碼略有不同，提交到伺服器叢集時需要使用 StormSubmitter 進行提交。主要程式碼如下：
為了結構清晰，這裡新建 ClusterWordCountApp 類來演示叢集模式的提交。實際開發中可以將兩種模式的程式碼寫在同一個類中，通過外部傳參來決定啟動何種模式。
public class ClusterWordCountApp {
public static void main(String[] args) {
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("DataSourceSpout", new DataSourceSpout());
// 指明將 DataSourceSpout 的資料傳送到 SplitBolt 中處理
builder.setBolt("SplitBolt", new SplitBolt()).shuffleGrouping("DataSourceSpout");
// 指明將 SplitBolt 的資料傳送到 CountBolt 中 處理
builder.setBolt("CountBolt", new CountBolt()).shuffleGrouping("SplitBolt");
// 使用 StormSubmitter 提交 Topology 到伺服器叢集
try {
StormSubmitter.submitTopology("ClusterWordCountApp", new Config(), builder.createTopology());
} catch (AlreadyAliveException | InvalidTopologyException | AuthorizationException e) {
e.printStackTrace();
}
}
}6.2 打包上傳
打包後上傳到伺服器任意位置，這裡我打包後的名稱為 storm-word-count-1.0.jar
# mvn clean package -Dmaven.test.skip=true6.3 提交Topology
使用以下命令提交 Topology 到叢集：
# 命令格式: storm jar jar包位置 主類的全路徑 ...可選傳參
storm jar /usr/appjar/storm-word-count-1.0.jar com.heibaiying.wordcount.ClusterWordCountApp出現 successfully 則代表提交成功：
6.4 檢視Topology與停止Topology（命令列方式）
# 檢視所有Topology
storm list
# 停止 storm kill topology-name [-w wait-time-secs]
storm kill ClusterWordCountApp -w 3
6.5 檢視Topology與停止Topology（介面方式）
使用 UI 介面同樣也可進行停止操作，進入 WEB UI 介面（8080 埠），在 Topology Summary 中點選對應 Topology 即可進入詳情頁面進行操作。
七、關於專案打包的擴充套件說明
mvn package的侷限性
在上面的步驟中，我們沒有在 POM 中配置任何外掛，就直接使用 mvn package 進行專案打包，這對於沒有使用外部依賴包的專案是可行的。但如果專案中使用了第三方 JAR 包，就會出現問題，因為 package 打包後的 JAR 中是不含有依賴包的，如果此時你提交到伺服器上執行，就會出現找不到第三方依賴的異常。
這時候可能大家會有疑惑，在我們的專案中不是使用了 storm-core 這個依賴嗎？其實上面之所以我們能執行成功，是因為在 Storm 的叢集環境中提供了這個 JAR 包，在安裝目錄的 lib 目錄下：
為了說明這個問題我在 Maven 中引入了一個第三方的 JAR 包，並修改產生資料的方法：
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
<version>3.8.1</version>
</dependency>StringUtils.join() 這個方法在 commons.lang3 和 storm-core 中都有，原來的程式碼無需任何更改，只需要在 import 時指明使用 commons.lang3。
import org.apache.commons.lang3.StringUtils;
private String productData() {
Collections.shuffle(list);
Random random = new Random();
int endIndex = random.nextInt(list.size()) % (list.size()) + 1;
return StringUtils.join(list.toArray(), "\t", 0, endIndex);
}此時直接使用 mvn clean package 打包執行，就會丟擲下圖的異常。因此這種直接打包的方式並不適用於實際的開發，因為實際開發中通常都是需要第三方的 JAR 包。
想把依賴包一併打入最後的 JAR 中，maven 提供了兩個外掛來實現，分別是 maven-assembly-plugin 和 maven-shade-plugin。鑑於本篇文章篇幅已經比較長，且關於 Storm 打包還有很多需要說明的地方，所以關於 Storm 的打包方式單獨整理至下一篇文章：
Storm 三種打包方式對比分析
參考資料
- Running Topologies on a Production Cluster
- Pre-defined Descriptor Files
更多大資料系列文章可以參見 GitHub 開源專案： 大資料入門指南