一個Apache Storm v2.0流計算入門項目的開發、測試和運行（IDEA/Maven）

關于流計算框架Apache Storm最新版的安裝，可以參考：

流計算框架-最新版Apache Storm v2.0單機模式安裝詳細步驟

流計算框架Apache Storm核心概念、架構設計

一、基于IDEA/Maven創建一個Storm應用

應用名稱：firststorm

二、添加storm-client的Maven jar包依賴

storm-client 依賴包信息，添加到項目的pom.xml文件中。

<dependency>

<groupId>org.apache.storm</groupId>

<artifactId>storm-client</artifactId>

<version>2.0.0</version>

</dependency>

maven會自動下載相關依賴并放到Maven Dependencies下，這些jar包可以點擊下拉查看，并且會自動添加到項目classpath中，作為編譯使用，等jar包全部下載完畢，現在開始編寫具體的計算邏輯了，在這個項目中我們把所有的類都建立在包com.rickie.bigdata.firststorm下。

storm提供了兩種運行模式：本地模式（Local Mode）和分布式模式。本地模式針對開發調試storm topologies非常有用。

因為多數程序開發者都是使用windows系統進行程序開發，如果在本機不安裝storm環境的情況下，如何在本地開發、調試storm程序呢？你可以參考本文提供的解決方案。

如下來自Storm 官方文檔：

http://storm.apache.org/releases/2.0.0-SNAPSHOT/Local-mode.html

Local Mode

Local mode simulates a Storm cluster in process and is useful for developing and testing topologies. Running topologies in local mode is similar to running topologies on a cluster.

To run a topology in local mode you have two options. The most common option is to run your topology with storm local instead of storm jar.

This will bring up a local simulated cluster and force all interactions with nimbus to go through the simulated cluster instead of going to a separate process.

If you want to do some automated testing but without actually launching a storm cluster you can use the same classes internally that storm local does.

To do this you first need to pull in the dependencies needed to access these classes. For the java API you should depend on storm-server as a test dependency.

To create an in-process cluster, simply use the LocalCluster class.

如上文所述，使用本地模式（Local Mode），需要先引入storm-server 依賴包。

<dependency>

<groupId>org.apache.storm</groupId>

<artifactId>storm-server</artifactId>

<version>2.0.0</version>

<scope>test</scope>

</dependency>

引入的storm-server 依賴包。

在本地模式上運行topology類似在一個集群上運行topology。

創建一個本地集群，大致代碼如下所示：

• import org.apache.storm.LocalCluster;
• LocalCluster cluster = new LocalCluster();
• 提交集群使用submitTopology，
• 殺死集群使用killTopology
• 關閉一個本地集群使用cluster.shutdown();

完整代碼可以參考下面。

三、Storm 應用的代碼邏輯開發

（1）首先建立RandomSpout類作為數據源，并且繼承于父類BaseRichSpout，確定后可以看到系統自動補全3個方法：nextTuple,open和declareOutputFields。

我們現在就需要重寫這3個方法，open方法是數據源的初始化，nextTuple的作用是把Tuple發送至下游，declareOutputFields用來定義輸出字段，下面我們手動分配一個數組，并且隨機取里面的元素，代碼如下：

package com.rickie.bigdata;

import org.apache.storm.spout.SpoutOutputCollector;

import org.apache.storm.task.TopologyContext;

import org.apache.storm.topology.OutputFieldsDeclarer;

import org.apache.storm.topology.base.BaseRichSpout;

import org.apache.storm.tuple.Fields;

import org.apache.storm.tuple.Values;

import java.util.Map;

import java.util.Random;

public class RandomSpout extends BaseRichSpout {

private SpoutOutputCollector collector;

private static String[] words = {"Rickie", "Hadoop", "MapReduce", "Storm", "Spark", "Spark Streaming", "Flink"};

@Override

public void open(Map<String, Object> map, TopologyContext topologyContext,

SpoutOutputCollector spoutOutputCollector) {

this.collector = spoutOutputCollector;

}

@Override

public void nextTuple() {

String word = words[new Random().nextInt(words.length)];

collector.emit(new Values(word));

}

@Override

public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

outputFieldsDeclarer.declare(new Fields("randomString"));

}

}

（2）然后新建一個類SenqueceBolt，繼承于BaseBasicBolt類，并且重寫方法execute和declareOutputFields。

這個類就是用于執行具體的作業，準確的說是execute方法用來執行相關的計算，這里只是簡單的輸出，代碼如下：

package com.rickie.bigdata;

import org.apache.storm.topology.BasicOutputCollector;

import org.apache.storm.topology.OutputFieldsDeclarer;

import org.apache.storm.topology.base.BaseBasicBolt;

import org.apache.storm.tuple.Tuple;

public class SequenceBolt extends BaseBasicBolt {

@Override

public void execute(Tuple tuple, BasicOutputCollector basicOutputCollector) {

String word = (String) tuple.getValue(0);

String out = "Hello " + word +"!";

System.out.println(out);

}

@Override

public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

}

}

（3）最后建立一個類FirstStorm。

這個類是主類，在main方法中定義Topology，并且綜合設置Spout和Bolt，從而調用其中的方法，這里流式計算時間設置為30s，代碼如下：

package com.rickie.bigdata;

import org.apache.storm.Config;

import org.apache.storm.LocalCluster;

import org.apache.storm.StormSubmitter;

import org.apache.storm.topology.TopologyBuilder;

import org.apache.storm.utils.Utils;

public class FirstStorm

{

public static void main( String[] args ) throws Exception {

TopologyBuilder builder = new TopologyBuilder();

builder.setSpout("spout", new RandomSpout());

builder.setBolt("bolt", new SequenceBolt()).shuffleGrouping("spout");

Config conf = new Config();

conf.setDebug(false);

String name = "firststorm";

if(args != null && args.length >0){

name = args[0];

conf.setNumWorkers(3);

try {

StormSubmitter.submitTopology(name, conf, builder.createTopology());

} catch (Exception e) {

e.printStackTrace();

}

} else {

try(LocalCluster cluster = new LocalCluster()) {

cluster.submitTopology("firststorm", conf, builder.createTopology());

Utils.sleep(10000);

cluster.killTopology("firststorm");

cluster.shutdown();

}

}

System.out.println( "Well done!" );

}

}

四、運行調試Storm應用

可以用本地模式運行，在IDEA中直接運行即可，方便開發調試。

在Console可以看到如下輸出信息：

接下來我們將這個項目放到Storm服務器集群中運行。

storm jar 命令用于啟動一個Topology。

（1）以本地模式（Local Mode）運行

storm local firststorm-1.0-SNAPSHOT.jar com.rickie.bigdata.FirstStorm

運行結果和之前IDEA本地模式運行輸出類似。

（2）以分布式模式運行

storm jar firststorm-1.0-SNAPSHOT.jar com.rickie.bigdata.FirstStorm RickieStorm

可以查看worker日志，看到SequenceBolt 線程輸出的信息。只有kill 這個topology，日志輸出信息才會終止。

在Storm架構中，Topology代表的并不是確定的作業，而是持續的計算過程，在確定的業務邏輯處理框架下，輸入數據源源不斷地進入系統，經過流式處理后以較低的延遲產生輸出。如果不主動結束這個Topology或者關閉Storm集群，那么數據處理的過程就會持續地進行下去。

另外需要注意的是：由于在分布式模式下運行，worker工作在獨立的進程中，因此無法直接在storm jar命令行輸出窗口，看到上述SequenceBolt組件的輸出信息。

（3）storm list 查看正在運行的topologies和它們的狀態

storm list

也可以通過訪問http://192.168.56.103:8080/ storm server，查看并操作正在運行的 topology。

（4）kill正在運行的topology

storm kill [storm name]

運行“storm kill”這個命令，僅僅只是調用Nimbus的Thirft接口去kill掉相對應的Topology。

Nimbus接受到kill命令，會將”kill”事務應用到topology上，修改Topology的狀態為”killed”以及將“remove”事件列入到未來幾秒鐘的計劃中，即未來幾秒后會觸發remove時間。這里的kill實際上停止相關的Worker。

默認kill的等待時間是Topology消息的超時時間，但是可以通過storm kill命令中的-w標志對其進行重寫，設置了以上參數之后，topology會在指定的等待時間停止運行。這樣給了Topology一個機會在shutdown workers之后完成當前沒有處理完成的任務；刪除Topology以及清理zookeeper中的分配信息和靜態信息；清理存儲在本地的心跳dir和jar/configs。

現在，第一個Storm入門項目的開發和測試運行都完畢了，更復雜的流計算邏輯模式也基本相同，主要就是Maven項目中出現了更復雜的模塊和調用，整個運行的流程其實都是差不多的。恭喜你，現在算是步入Storm流式計算的殿堂的大門了。

五、Storm架構是如何解決Hadoop架構瓶頸的？

• Storm的Topology只需初始化一次。在將Topology提交到Storm集群的時候，集群會針對該Topology做一次初始化的工作。此后，在Topology運行過程中，對于輸入數據而言，是沒有計算框架初始化耗時的，有效避免了計算框架初始化的時間損耗。
• Storm使用Netty作為底層的消息隊列來傳遞消息，保證消息能夠得到快速的處理。
• 同時Storm采用內存計算模式，無需借助文件存儲，直接通過網絡直傳中間計算結果，避免了組件之間傳輸數據的大量時間損耗。

六、Apache Storm中的核心概念

• Topology：一個實時計算任務被稱作為Topology，包含Spout和Bolt。

• Tuple：數據模型，代表處理單元，可以包含多個Field，K/V的Map。
• Worker：一個topology可能會在一個或者多個worker(工作進程)里面執行，每個worker是一個物理JVM并且執行整個topology的一部分。比如，對于并行度是300的topology來說，如果我們使用50個工作進程worker來執行，那么每個工作進程會處理其中的6個tasks。Storm會盡量均勻的工作分配給所有的worker，setBolt 的最后一個參數是你想為bolts的并行量。
• Spouts

1. 消息源Spout是Storm里面一個topology里面的消息生產者。一般來說消息源會從一個外部源讀取數據并且向topology里面發出消息：tuple。Spout可以是可靠的也可以是不可靠的，如果這個tuple沒有被storm成功處理,可靠的消息源spouts可以重新發射一個tuple，但是不可靠的消息源spouts一旦發出一個tuple就不能重發了。
2. 消息源可以發射多條消息流stream。使用OutputFieldsDeclarer。declareStream來定義多個stream,然后使用SpoutOutputCollector來發射指定的stream。代碼上是這樣的:collector.emit(new Values(str));
3. Spout類里面最重要的方法是nextTuple。要么發射一個新的tuple到topology里面或者簡單的返回如果已經沒有新的tuple。要注意的是nextTuple方法不能阻塞，因為storm在同一個線程上面調用所有消息源spout的方法。另外兩個比較重要的spout方法是ack和fail。storm在檢測到一個tuple被整個topology成功處理的時候調用ack，否則調用fail。storm只對可靠的spout調用ack和fail。

• Bolts

1. 所有的消息處理邏輯被封裝在bolts里面。Bolts可以做很多事情：過濾，聚合，查詢數據庫等等。
2. Bolts可以簡單的做消息流的傳遞(來一個元組,調用一次execute)。復雜的消息流處理往往需要很多步驟，從而也就需要經過很多bolts。比如算出一堆圖片里面被轉發最多的圖片就至少需要兩步：第一步算出每個圖片的轉發數量，第二步找出轉發最多的前10個圖片。(如果要把這個過程做得更具有擴展性那么可能需要更多的步驟)。
3. Bolts可以發射多條消息流, 使用OutputFieldsDeclarer.declareStream定義stream，使用OutputCollector.emit來選擇要發射的stream。
4. Bolts的主要方法是execute，它以一個tuple作為輸入，bolts使用OutputCollector來發射tuple(spout使用SpoutOutputCollector來發射指定的stream)，bolts必須要為它處理的每一個tuple調用OutputCollector的ack方法，以通知Storm這個tuple被處理完成了，從而通知這個tuple的發射者spouts。一般的流程是: bolts處理一個輸入tuple, 發射0個或者多個tuple, 然后調用ack通知storm自己已經處理過這個tuple了。storm提供了一個IBasicBolt會自動調用ack。