?打成jar包放在主節(jié)點(diǎn)上去運(yùn)行.

1 import java.util.Map; 2 3 import backtype.storm.Config; 4 import backtype.storm.StormSubmitter; 5 import backtype.storm.generated.AlreadyAliveException; 6 import backtype.storm.generated.InvalidTopologyException; 7 import backtype.storm.spout.SpoutOutputCollector; 8 import backtype.storm.task.OutputCollector; 9 import backtype.storm.task.TopologyContext; 10 import backtype.storm.topology.OutputFieldsDeclarer; 11 import backtype.storm.topology.TopologyBuilder; 12 import backtype.storm.topology.base.BaseRichBolt; 13 import backtype.storm.topology.base.BaseRichSpout; 14 import backtype.storm.tuple.Fields; 15 import backtype.storm.tuple.Tuple; 16 import backtype.storm.tuple.Values; 17 import backtype.storm.utils.Utils; 18 19 /** 20 * 在集群運(yùn)行 21 * 數(shù)字累加求和 22 * 先添加storm依賴 23 * 24 * @author Administrator 25 * 26 */ 27 public class StormTopologySum { 28 29 30 /** 31 * spout需要繼承baserichspout，實現(xiàn)未實現(xiàn)的方法 32 * @author Administrator 33 * 34 */ 35 public static class MySpout extends BaseRichSpout{ 36 private Map conf; 37 private TopologyContext context; 38 private SpoutOutputCollector collector; 39 40 /** 41 * 初始化方法，只會執(zhí)行一次 42 * 在這里面可以寫一個初始化的代碼 43 * Map conf：其實里面保存的是topology的一些配置信息 44 * TopologyContext context：topology的上下文，類似于servletcontext 45 * SpoutOutputCollector collector：發(fā)射器，負(fù)責(zé)向外發(fā)射數(shù)據(jù)(tuple) 46 */ 47 @Override 48 public void open(Map conf, TopologyContext context, 49 SpoutOutputCollector collector) { 50 this.conf = conf; 51 this.context = context; 52 this.collector = collector; 53 } 54 55 int num = 1; 56 /** 57 * 這個方法是spout中最重要的方法， 58 * 這個方法會被storm框架循環(huán)調(diào)用，可以理解為這個方法是在一個while循環(huán)之內(nèi) 59 * 每調(diào)用一次，會向外發(fā)射一條數(shù)據(jù) 60 */ 61 @Override 62 public void nextTuple() { 63 System.out.println("spout發(fā)射："+num); 64 //把數(shù)據(jù)封裝到values中，稱為一個tuple，發(fā)射出去 65 this.collector.emit(new Values(num++)); 66 Utils.sleep(1000); 67 } 68 69 /** 70 * 聲明輸出字段 71 */ 72 @Override 73 public void declareOutputFields(OutputFieldsDeclarer declarer) { 74 //給values中的數(shù)據(jù)起個名字，方便后面的bolt從這個values中取數(shù)據(jù) 75 //fields中定義的參數(shù)和values中傳遞的數(shù)值是一一對應(yīng)的 76 declarer.declare(new Fields("num")); 77 } 78 79 } 80 81 82 /** 83 * 自定義bolt需要實現(xiàn)baserichbolt 84 * @author Administrator 85 * 86 */ 87 public static class MyBolt extends BaseRichBolt{ 88 private Map stormConf; 89 private TopologyContext context; 90 private OutputCollector collector; 91 92 /** 93 * 和spout中的open方法意義一樣 94 */ 95 @Override 96 public void prepare(Map stormConf, TopologyContext context, 97 OutputCollector collector) { 98 this.stormConf = stormConf; 99 this.context = context; 100 this.collector = collector; 101 } 102 103 int sum = 0; 104 /** 105 * 是bolt中最重要的方法，當(dāng)spout發(fā)射一個tuple出來，execute也會被調(diào)用，需要對spout發(fā)射出來的tuple進(jìn)行處理 106 */ 107 @Override 108 public void execute(Tuple input) { 109 //input.getInteger(0);//也可以根據(jù)角標(biāo)獲取tuple中的數(shù)據(jù) 110 Integer value = input.getIntegerByField("num"); 111 sum+=value; 112 System.out.println("和："+sum); 113 } 114 115 /** 116 * 聲明輸出字段 117 */ 118 @Override 119 public void declareOutputFields(OutputFieldsDeclarer declarer) { 120 //在這沒必要定義了，因為execute方法中沒有向外發(fā)射tuple，所以就不需要聲明了。 121 //如果nextTuple或者execute方法中向外發(fā)射了tuple，那么declareOutputFields必須要聲明，否則不需要聲明 122 } 123 124 } 125 /** 126 * 注意：在組裝topology的時候，組件的id在定義的時候，名稱不能以__開頭。__是系統(tǒng)保留的 127 * @param args 128 */ 129 public static void main(String[] args) { 130 //組裝topology 131 TopologyBuilder topologyBuilder = new TopologyBuilder(); 132 topologyBuilder.setSpout("spout1", new MySpout()); 133 //.shuffleGrouping("spout1"); 表示讓MyBolt接收MySpout發(fā)射出來的tuple 134 topologyBuilder.setBolt("bolt1", new MyBolt()).setNumTasks(2).shuffleGrouping("spout1"); 135 136 //創(chuàng)建本地storm集群 137 /*LocalCluster localCluster = new LocalCluster(); 138 localCluster.submitTopology("sumTopology", new Config(), topologyBuilder.createTopology());*/ 139 140 //在集群運(yùn)行 141 String simpleName = StormTopologySum.class.getSimpleName(); 142 try { 143 Config stormConf = new Config(); 144 //stormConf.setNumWorkers(2); 145 StormSubmitter.submitTopology(simpleName, stormConf, topologyBuilder.createTopology()); 146 } catch (AlreadyAliveException e) { 147 e.printStackTrace(); 148 } catch (InvalidTopologyException e) { 149 e.printStackTrace(); 150 } 151 } 152 }

?

轉(zhuǎn)載于:https://www.cnblogs.com/DreamDrive/p/5786198.html

與50位技術(shù)專家面對面20年技術(shù)見證，附贈技術(shù)全景圖

總結(jié)

以上是生活随笔為你收集整理的Storm累计求和Demo并且在集群上运行的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。