建站
咨詢
售后
建站咨詢
新聞中心
Storm是一個開源的分布式實時計算系統(tǒng)，它可以處理大量的數(shù)據(jù)流并進行實時分析，在實際應(yīng)用中，單詞計數(shù)是一種常見的需求，可以通過Storm來實現(xiàn)，下面將詳細介紹如何使用Storm實現(xiàn)單詞計數(shù)。
創(chuàng)新互聯(lián)長期為上1000+客戶提供的網(wǎng)站建設(shè)服務(wù)，團隊從業(yè)經(jīng)驗10年，關(guān)注不同地域、不同群體，并針對不同對象提供差異化的產(chǎn)品和服務(wù)；打造開放共贏平臺，與合作伙伴共同營造健康的互聯(lián)網(wǎng)生態(tài)環(huán)境。為烏審企業(yè)提供專業(yè)的網(wǎng)站設(shè)計、成都網(wǎng)站建設(shè)，烏審網(wǎng)站改版等技術(shù)服務(wù)。擁有十多年豐富建站經(jīng)驗和眾多成功案例,為您定制開發(fā)。
我們需要創(chuàng)建一個Storm拓撲結(jié)構(gòu)，Storm拓撲由一個或多個Spouts（數(shù)據(jù)源）和Bolts（數(shù)據(jù)處理單元）組成，在這個例子中，我們將使用一個簡單的Spout來生成單詞流，然后使用一個Bolt來計算每個單詞的出現(xiàn)次數(shù)。
1. 創(chuàng)建Spout：Spout是Storm拓撲的數(shù)據(jù)源，它負責(zé)生成數(shù)據(jù)流，在這個例子中，我們可以使用隨機數(shù)生成器來模擬單詞流，創(chuàng)建一個名為WordSpout的Java類，繼承自BaseRichSpout類，重寫nextTuple方法，每次調(diào)用時生成一個隨機單詞作為輸出。
import backtype.storm.spout.BaseRichSpout;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import java.util.Map;
import java.util.Random;
public class WordSpout extends BaseRichSpout {
private SpoutOutputCollector collector;
private Random random;
@Override
public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
this.collector = collector;
this.random = new Random();
}
@Override
public void nextTuple() {
String word = "word" + random.nextInt(100);
collector.emit(new Values(word));
}
}
2. 創(chuàng)建Bolt：Bolt是Storm拓撲的數(shù)據(jù)處理單元，它負責(zé)對數(shù)據(jù)流進行處理，在這個例子中，我們可以使用HashMap來存儲每個單詞的出現(xiàn)次數(shù)，創(chuàng)建一個名為WordCounterBolt的Java類，繼承自BaseRichBolt類，重寫execute方法，每次接收到一個單詞時，將其出現(xiàn)次數(shù)加一，使用collector將結(jié)果發(fā)送出去。
import backtype.storm.bolt.BaseRichBolt;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import java.util.HashMap;
import java.util.Map;
import java.util.Iterator;
import java.util.Map.Entry;
public class WordCounterBolt extends BaseRichBolt {
private OutputCollector collector;
private Map wordCounts;
@Override
public void prepare(Map conf, TopologyContext context, OutputCollector collector) {
this.collector = collector;
this.wordCounts = new HashMap<>();
}
@Override
public void execute(Tuple tuple) {
String word = tuple.getStringByField("word");
int count = wordCounts.containsKey(word) ? wordCounts.get(word) + 1 : 1;
wordCounts.put(word, count);
collector.emit(new Values(word, count));
}
}
3. 配置拓撲：接下來，我們需要配置Storm拓撲，創(chuàng)建一個名為WordCountTopology的Java類，繼承自BaseMainClass類，重寫buildTopology方法，設(shè)置Spout和Bolt的配置參數(shù)，啟動拓撲。
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import backtype.storm.utils.Utils;
import storm_wordcount_example.*; // 導(dǎo)入自定義的Spout和Bolt類
public class WordCountTopology {
public static void main(String[] args) throws Exception {
LocalCluster cluster = new LocalCluster();
cluster.submitTopology("word-count", buildTopology());
Utils.sleep(10000); // 等待10秒后關(guān)閉集群
cluster.shutdown();
}
private static TopologyBuilder buildTopology() {
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("word-spout", new WordSpout(), 5); // 設(shè)置Spout的并發(fā)度為5
builder.setBolt("word-counter", new WordCounterBolt(), 5).shuffleGrouping("word-spout"); // 設(shè)置Bolt的并發(fā)度為5，并指定分組策略為隨機分組（shuffle grouping）
return builder;
}
}
4. 運行拓撲：運行WordCountTopology類，觀察單詞計數(shù)的結(jié)果，在Storm UI中，可以看到每個單詞的出現(xiàn)次數(shù)以及總計數(shù)，還可以查看拓撲的狀態(tài)、任務(wù)分配等信息。
當(dāng)前題目：storm怎么記
文章分享：http://www.5511xx.com/article/djoihgh.html
