Apache Storm v2.0入門項目的開發(fā)、測試和運行（IDEA/Maven）

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

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

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

流計算框架Apache Storm核心概念、架構(gòu)設(shè)計

一、基于IDEA/Maven創(chuàng)建一個Storm應(yīng)用

應(yīng)用名稱：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會自動下載相關(guān)依賴并放到Maven Dependencies下，這些jar包可以點擊下拉查看，并且會自動添加到項目classpath中，作為編譯使用，等jar包全部下載完畢，現(xiàn)在開始編寫具體的計算邏輯了，在這個項目中我們把所有的類都建立在包com.rickie.bigdata.firststorm下。

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

因為多數(shù)程序開發(fā)者都是使用windows系統(tǒng)進行程序開發(fā)，如果在本機不安裝storm環(huán)境的情況下，如何在本地開發(fā)、調(diào)試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。

創(chuàng)建一個本地集群，大致代碼如下所示：

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

完整代碼可以參考下面。

三、Storm 應(yīng)用的代碼邏輯開發(fā)

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

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

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。

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

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，并且綜合設(shè)置Spout和Bolt，從而調(diào)用其中的方法，這里流式計算時間設(shè)置為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!" );
}
}

四、運行調(diào)試Storm應(yīng)用

可以用本地模式運行，在IDEA中直接運行即可，方便開發(fā)調(diào)試。

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

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

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

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

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

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

（2）以分布式模式運行

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

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

在Storm架構(gòu)中，Topology代表的并不是確定的作業(yè)，而是持續(xù)的計算過程，在確定的業(yè)務(wù)邏輯處理框架下，輸入數(shù)據(jù)源源不斷地進入系統(tǒng)，經(jīng)過流式處理后以較低的延遲產(chǎn)生輸出。如果不主動結(jié)束這個Topology或者關(guān)閉Storm集群，那么數(shù)據(jù)處理的過程就會持續(xù)地進行下去。

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

（3）storm list 查看正在運行的topologies和它們的狀態(tài)

storm list

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

（4）kill正在運行的topology

storm kill [storm name]

運行“storm kill”這個命令，僅僅只是調(diào)用Nimbus的Thirft接口去kill掉相對應(yīng)的Topology。

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

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

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

五、Storm架構(gòu)是如何解決Hadoop架構(gòu)瓶頸的？

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

六、Apache Storm中的核心概念

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

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

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

Bolts

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

程鏈接：https://www.itwangzi.cn/4907.html

Storm 實現(xiàn)了低延遲，還做不到高吞吐，也不能在故障發(fā)生時準確地處理計算狀態(tài)；Spark Streaming通過采用微批處理方法實現(xiàn)了高吞吐和容錯性，但是犧牲了低延遲和實時處理能力，也不能使窗口與自然時間相匹配，并且表現(xiàn)力欠佳。而flink就是目前為止的最佳答案。

我們在選擇一個新的技術(shù)框架的時候，首先考慮的是他的應(yīng)用場景，再牛逼的框架沒有應(yīng)用場景也是一無是處，當然牛逼的框架大多都是基于某一個或者某一類應(yīng)用場景而產(chǎn)生，而flink主要應(yīng)用于以下三個場景：

Flink 主要應(yīng)用場景有三類：

1.Event-driven Applications【事件驅(qū)動】

2.Data Analytics Applications【分析】

3.Data Pipeline Applications【管道式ETL】

Event-driven Applications

上圖包含兩塊：

Traditional transaction Application（傳統(tǒng)事務(wù)應(yīng)用）

Event-driven Applications（事件驅(qū)動應(yīng)用）

Traditional transaction Application執(zhí)行流程：

比如點擊流Events可以通過Application寫入Transaction DB（數(shù)據(jù)庫），同時也可以通過Application從Transaction DB將數(shù)據(jù)讀出，并進行處理，當處理結(jié)果達到一個預(yù)警值就會觸發(fā)一個Action動作，這種方式一般為事后諸葛亮。

Event-driven Applications執(zhí)行流程：

比如采集的數(shù)據(jù)Events可以不斷的放入消息隊列，F(xiàn)link應(yīng)用會不斷ingest（消費）消息隊列中的數(shù)據(jù)，F(xiàn)link 應(yīng)用內(nèi)部維護著一段時間的數(shù)據(jù)（state），隔一段時間會將數(shù)據(jù)持久化存儲（Persistent sstorage），防止Flink應(yīng)用死掉。Flink應(yīng)用每接受一條數(shù)據(jù)，就會處理一條數(shù)據(jù)，處理之后就會觸發(fā)（trigger）一個動作（Action），同時也可以將處理結(jié)果寫入外部消息隊列中，其他Flink應(yīng)用再消費。

典型的事件驅(qū)動類應(yīng)用：

1.欺詐檢測(Fraud detection)

2.異常檢測(Anomaly detection)

3.基于規(guī)則的告警(Rule-based alerting)

4.業(yè)務(wù)流程監(jiān)控(Business process monitoring)

5.Web應(yīng)用程序(社交網(wǎng)絡(luò))

Data Analytics Applications

Data Analytics Applications包含Batch analytics（批處理分析）和Streaming analytics（流處理分析）。

Batch analytics可以理解為周期性查詢：

比如Flink應(yīng)用凌晨從Recorded Events中讀取昨天的數(shù)據(jù)，然后做周期查詢運算，最后將數(shù)據(jù)寫入Database或者HDFS，或者直接將數(shù)據(jù)生成報表供公司上層領(lǐng)導(dǎo)決策使用。

Streaming analytics可以理解為連續(xù)性查詢：

比如實時展示雙十一天貓銷售GMV，用戶下單數(shù)據(jù)需要實時寫入消息隊列，F(xiàn)link 應(yīng)用源源不斷讀取數(shù)據(jù)做實時計算，然后不斷的將數(shù)據(jù)更新至Database或者K-VStore，最后做大屏實時展示。

Data Pipeline Applications

Data Pipeline Applications包含Periodic （周期性）ETL和Data Pipeline（管道）

Periodic ETL：比如每天凌晨周期性的啟動一個Flink ETL Job，讀取傳統(tǒng)數(shù)據(jù)庫中的數(shù)據(jù)，然后做ETL，最后寫入數(shù)據(jù)庫和文件系統(tǒng)。

Data Pipeline：比如啟動一個Flink 實時應(yīng)用，數(shù)據(jù)源（比如數(shù)據(jù)庫、Kafka）中的數(shù)據(jù)不斷的通過Flink Data Pipeline流入或者追加到數(shù)據(jù)倉庫（數(shù)據(jù)庫或者文件系統(tǒng)），或者Kafka消息隊列。

阿里Flink應(yīng)用場景

阿里在Flink的應(yīng)用主要包含四個模塊：實時監(jiān)控、實時報表、流數(shù)據(jù)分析和實時倉庫。

實時監(jiān)控：

用戶行為預(yù)警、app crash 預(yù)警、服務(wù)器攻擊預(yù)警
對用戶行為或者相關(guān)事件進行實時監(jiān)測和分析，基于風控規(guī)則進行預(yù)警

實時報表：

雙11、雙12等活動直播大屏
對外數(shù)據(jù)產(chǎn)品：生意參謀等
數(shù)據(jù)化運營

流數(shù)據(jù)分析：

實時計算相關(guān)指標反饋及時調(diào)整決策
內(nèi)容投放、無線智能推送、實時個性化推薦等

實時倉庫：

數(shù)據(jù)實時清洗、歸并、結(jié)構(gòu)化
數(shù)倉的補充和優(yōu)化

欺詐檢測

背景：

假設(shè)你是一個電商公司，經(jīng)常搞運營活動，但收效甚微，經(jīng)過細致排查，發(fā)現(xiàn)原來是羊毛黨在薅平臺的羊毛，把補給用戶的補貼都薅走了，錢花了不少，效果卻沒達到。怎么辦呢？

你可以做一個實時的異常檢測系統(tǒng)，監(jiān)控用戶的高危行為，及時發(fā)現(xiàn)高危行為并采取措施，降低損失。

系統(tǒng)流程：

1.用戶的行為經(jīng)由app 上報或web日志記錄下來，發(fā)送到一個消息隊列里去；

2.然后流計算訂閱消息隊列，過濾出感興趣的行為，比如：購買、領(lǐng)券、瀏覽等；

3.流計算把這個行為特征化；

4.流計算通過UDF調(diào)用外部一個風險模型，判斷這次行為是否有問題（單次行為）；

5.流計算里通過CEP功能，跨多條記錄分析用戶行為（比如用戶先做了a，又做了b，又做了3次c），整體識別是否有風險；

6.綜合風險模型和CEP的結(jié)果，產(chǎn)出預(yù)警信息。

《大數(shù)據(jù)和人工智能交流》頭條號向廣大初學者新增C 、Java 、Python 、Scala、javascript 等目前流行的計算機、大數(shù)據(jù)編程語言，希望大家以后關(guān)注本頭條號更多的內(nèi)容。

一、搭建zookeeper集群

注意：為了大家學習的方便，這里在一臺機器上搭建zookeeper集群，在一個機器搭建集群和在多臺機器搭建集群原理是相同的。搭建單節(jié)點（一臺主機放3個服務(wù)【1個leader，2個flower】）zk集群

在/home建立zk1、zk2、zk3三個目錄,3個目錄搭建過程類似。

（一）搭建過程

【1】搭建節(jié)點1

(1)在zk1建立zkdata ，在zkdata下存放myid文件

[root@node1 zkdata]# more myid

(2) tar -zxvf zookeeper-3.4.5.tar.gz

[root@node1 zk1]# ls

zkdata zookeeper-3.4.5 zookeeper-3.4.5.tar.gz

(3)修改配置文件

[root@node1 zk1]# cd zookeeper-3.4.5

[root@node1 zookeeper-3.4.5]# cd conf

[root@node1 zookeeper-3.4.5]# mv zoo_sample.cfg zoo.cfg

[root@node1 conf]# vi zoo.cfg

注意:修改的配置信息

dataDir=/tmp/zookeeper

# the port at which the clients will connect

clientPort=2181

dataDir=/home/zk1/zkdata

server.1=node1:2888:3888

server.2=node1:2889:3889

server.3=node1:2890:3890

【2】搭建節(jié)點2

(1)在zk1建立zkdata ，在zkdata下存放myid文件

[root@node1 zkdata]# more myid

(2) tar -zxvf zookeeper-3.4.5.tar.gz

[root@node1 zk1]# ls

zkdata zookeeper-3.4.5 zookeeper-3.4.5.tar.gz

(3)修改配置文件

[root@node1 zk1]# cd zookeeper-3.4.5

[root@node1 zookeeper-3.4.5]# cd conf

[root@node1 zookeeper-3.4.5]# mv zoo_sample.cfg zoo.cfg

[root@node1 conf]# vi zoo.cfg

注意:修改的配置信息

dataDir=/tmp/zookeeper

# the port at which the clients will connect

clientPort=2182

dataDir=/home/zk2/zkdata

server.1=node1:2888:3888

server.2=node1:2889:3889

server.3=node1:2890:3890

【3】搭建節(jié)點3

(1)在zk1建立zkdata ，在zkdata下存放myid文件

[root@node1 zkdata]# more myid

(2) tar -zxvf zookeeper-3.4.5.tar.gz

[root@node1 zk1]# ls

zkdata zookeeper-3.4.5 zookeeper-3.4.5.tar.gz

(3)修改配置文件

[root@node1 zk1]# cd zookeeper-3.4.5

[root@node1 zookeeper-3.4.5]# cd conf

[root@node1 zookeeper-3.4.5]# mv zoo_sample.cfg zoo.cfg

[root@node1 conf]# vi zoo.cfg

注意:修改的配置信息

dataDir=/tmp/zookeeper

# the port at which the clients will connect

clientPort=2183

dataDir=/home/zk3/zkdata

server.1=node1:2888:3888

server.2=node1:2889:3889

server.3=node1:2890:3890

（二）、啟動集群

1、啟動各個

[root@node1 bin]# ./zkServer.sh start

JMX enabled by default

Using config: /home/zk1/zookeeper-3.4.5/bin/../conf/zoo.cfg

Starting zookeeper ... ./zkServer.sh: line 103: [: /tmp/zookeeper: binary operator expected

STARTED

[root@node1 bin]#

2、在zk1、zk2、zk3各個目錄查看節(jié)點狀態(tài)

[root@node1 bin]# ./zkServer.sh status

JMX enabled by default

Using config: /home/zk1/zookeeper-3.4.5/bin/../conf/zoo.cfg

Mode: follower

二、搭建storm集群

（一）、準備工作

1、zk必須正常，3個節(jié)點聯(lián)網(wǎng)通信正常。我這里是3個節(jié)點：

192.168.100.1是主節(jié)點

192.168.100.2是從節(jié)點

192.168.100.3是從節(jié)點

將包上傳-->解壓 -->改名-->這里改為storm

[root@node1 storm001]# ls

apache-storm-0.9.3.tar.gz storm

2、修改配置文件

[root@node1 storm001]# cd storm

[root@node1 storm]# ls

bin conf examples lib logback NOTICE README.markdown SECURITY.md

CHANGELOG.md DISCLAIMER external LICENSE logs public RELEASE storm.jar

[root@node1 storm]# cd conf

[root@node1 conf]# ls

storm_env.ini storm.yaml

[root@node1 conf]# vi storm.yaml

storm.zookeeper.servers:

- "192.168.100.1"

nimbus.host: "192.168.100.1"

storm.local.dir: "/usr/local/storm/tmp"

supervisor.slots.ports:

- 6700

- 6701

- 6702

- 6703

解釋：

（1）storm.local.dir:存儲目錄

（2）nimbus.host: "192.168.100.1"代表選擇主機

（3）supervisor.slots.ports:對于每個supervisor機器，通過這項配置可以決定運行多少個worker進程在

這臺機器上，每個worker使用一個單獨的port來接受消息。supervisor并不會立即啟動這個4個worker進程，

當接收到分配的任務(wù)的時候會啟動，具體啟動多少個worker根據(jù)我們Topology在這個supervisor需要幾個worker

來確定

3、把storm目錄拷貝到其它2個節(jié)點

[root@node1 storm001]# ls

apache-storm-0.9.3.tar.gz storm

scp -r storm node2:/home/storm001

scp -r storm node3:/home/storm001

（二）、啟動集群

1、啟動主節(jié)點nimbus

[root@node1 bin]# ls

storm storm.cmd storm-config.cmd

[root@node1 bin]# ./storm nimbus

可以使用linux命令將其放到后臺./storm nimbus >/dev/null 2>&1 &

在主節(jié)點通過web前端訪問的進程

./storm ui

./storm ui >/dev/null 2>&1 &

2、啟動從節(jié)點node2

./storm supervisor

./storm supervisor >/dev/null 2>&1 &

從節(jié)點需要啟動日志進程

./storm logviewer

3、啟動從節(jié)點node3

./storm supervisor

./storm supervisor >/dev/null 2>&1 &

從節(jié)點需要啟動日志進程

./storm logviewer

4、進入storm的ui界面：

http://192.168.100.1:8080/index.html

5、將java開發(fā)的storm應(yīng)用打包上傳服務(wù)器

bin/storm jar storm.jar com.test.A001

《大數(shù)據(jù)和人工智能交流》的宗旨

1、將大數(shù)據(jù)和人工智能的專業(yè)數(shù)學：概率數(shù)理統(tǒng)計、線性代數(shù)、決策論、優(yōu)化論、博弈論等數(shù)學模型變得通俗易懂。

2、將大數(shù)據(jù)和人工智能的專業(yè)涉及到的數(shù)據(jù)結(jié)構(gòu)和算法：分類、聚類、回歸算法、概率等算法變得通俗易懂。

3、最新的高科技動態(tài)：數(shù)據(jù)采集方面的智能傳感器技術(shù)；醫(yī)療大數(shù)據(jù)智能決策分析；物聯(lián)網(wǎng)智慧城市等等。

根據(jù)初學者需要會有C語言、Java語言、Python語言、Scala函數(shù)式等目前主流計算機語言。

根據(jù)讀者的需要有和人工智能相關(guān)的計算機科學與技術(shù)、電子技術(shù)、芯片技術(shù)等基礎(chǔ)學科通俗易懂的文章。

在線咨詢

上一篇：JSPDF + html2canvas A4分頁截斷
下一篇：一行jQuery代碼搞定checkbox 全選和全不選

您的項目需求

*請認真填寫需求信息，我們會在24小時內(nèi)與您取得聯(lián)系。

整合營銷服務(wù)商

Apache Storm v2.0入門項目的開發(fā)、測試和運行（IDEA/Maven）

一、基于IDEA/Maven創(chuàng)建一個Storm應(yīng)用

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

三、Storm 應(yīng)用的代碼邏輯開發(fā)

四、運行調(diào)試Storm應(yīng)用

五、Storm架構(gòu)是如何解決Hadoop架構(gòu)瓶頸的？

六、Apache Storm中的核心概念

Event-driven Applications

Data Analytics Applications

Data Pipeline Applications

阿里Flink應(yīng)用場景

您的項目需求