这篇文章主要为大家展示了“storm如何配置使用”,内容简而易懂,条理清晰,希望能够帮助大家解决疑惑,下面让小编带领大家一起研究并学习一下“storm如何配置使用”这篇文章吧。
示例代码如下:
#storm.yaml 配置
#zookeeper
storm.zookeeper.servers:
- "bigdata01"
- "bigdata02"
- "bigdata03"
#本地存放数据的路径
storm.local.dir: "/apps/storm"
#nimbus master
nimbus.seeds: ["bigdata00"]
#workder端口
supervisor.slots.ports:
- 6700
- 6701
- 6702
- 6703
启动命令
bin/ nohup storm nimbus &
bin/ nohup storm supervisor &
bin/ nohup storm ui &
--------------------------------------------------------------------------------------
package com.hgs.storm;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.generated.AlreadyAliveException;
import org.apache.storm.generated.AuthorizationException;
import org.apache.storm.generated.InvalidTopologyException;
import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
public class StormWordCountTest {
public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException, AuthorizationException, InterruptedException {
TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("wordspout", new WordCountSpout(), 3);
builder.setBolt("splitword", (IRichBolt) new WordSpliteBolt(), 2).shuffleGrouping("wordspout");
//word 是splitword发出的字段,如第九十行
builder.setBolt("wordcount", new WordCountBolt(), 2).fieldsGrouping("splitword", new Fields("word"));
Config config = new Config();
config.setNumWorkers(2);
/* StormSubmitter.submitTopology("words-count", config, builder.createTopology());
if(args!=null && args.length>0) {
StormSubmitter.submitTopology(args[0], config, builder.createTopology());
}else {
LocalCluster cluster = new LocalCluster();
}*/
LocalCluster cluster = new LocalCluster();
cluster.submitTopology("words-count", config, builder.createTopology());
}
}
class WordCountSpout extends BaseRichSpout{
private static final long serialVersionUID = 1L;
//从open方法中的到collector,用于declareOutputFields 方法发出字段信息
SpoutOutputCollector collector = null;
@Override
public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
this.collector = collector;
}
@Override
public void nextTuple() {
collector.emit(new Values(" this is my first storm program so i hope it will success"));
}
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare( new Fields("message"));
}
}
class WordSpliteBolt extends BaseRichBolt{
private static final long serialVersionUID = 1L;
OutputCollector collector = null;
@Override
public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
this.collector = collector;
}
@Override
public void execute(Tuple input) {
String line = input.getString(0);
String[] words = line.split(" ");
for(String wd : words) {
collector.emit(new Values(wd ,1));
}
}
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("word","num"));
}
}
class WordCountBolt extends BaseRichBolt{
ConcurrentHashMap<String, Integer> wordsMap = new ConcurrentHashMap<String, Integer>();
private static final long serialVersionUID = 1L;
OutputCollector collector = null;
@Override
public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
this.collector = collector;
}
@Override
public void execute(Tuple input) {
String word = input.getString(0);
Integer num = input.getInteger(1);
if(wordsMap.containsKey(word)) {
wordsMap.put(word, wordsMap.get(word)+num);
}else {
wordsMap.put(word, num);
}
System.out.println(word +"----"+wordsMap.get(word));
}
@Override
public void declareOutputFields(OutputFieldsDeclarer declarer) {
}
}
以上是“storm如何配置使用”这篇文章的所有内容,感谢各位的阅读!相信大家都有了一定的了解,希望分享的内容对大家有所帮助,如果还想学习更多知识,欢迎关注亿速云行业资讯频道!
亿速云「云服务器」,即开即用、新一代英特尔至强铂金CPU、三副本存储NVMe SSD云盘,价格低至29元/月。点击查看>>
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。
原文链接:http://blog.itpub.net/31506529/viewspace-2214253/