Shulou(Shulou.com)05/31 Report--

This article mainly introduces "how to define storm startup class". In daily operation, I believe many people have doubts about how to define storm startup class. Xiaobian consulted all kinds of materials and sorted out simple and easy-to-use operation methods. I hope it will be helpful to answer the doubts about "how to define storm startup class". Next, please follow the editor to study!

There are three main instances that actually run Topology in a storm cluster: worker processes, threads, and tasks.

Each machine in the Storm cluster can run multiple worker processes, each worker process can create multiple threads, and each thread can execute multiple tasks.

Storm reliability: by giving a unique ID to the message tree, each message sent will send an ack or fail synchronously, which will consume the broadband of the network to a certain extent. If the reliability requirement is not high, the mode can be turned off by using different emit interfaces.

First, the definition of storm startup class.

Package com.cmsz.storm.trading.test;import backtype.storm.topology.TopologyBuilder;import backtype.storm.tuple.Fields;public class MainStorm {public static void main (String [] args) throws Exception {TopologyBuilder builder = new TopologyBuilder (); builder.setSpout ("A", new ASpout ()); builder.setBolt ("B", new BBolt ()) .shuffleGrouping ("A", "streamId_B") / / componentId and streamId builder.setBolt ("C", new CBolt ()) .shuffleGrouping ("A", "streamId_C"); / / componentId and streamId builder.setBolt ("D", new DBolt ()) .fieldsGrouping ("B", new Fields ("id")); / / componentId and streamId builder.setBolt ("E", new EBolt ()) .fieldsGrouping ("C", new Fields ("id")); Config conf = new Config () If (args! = null & & args.length > 0) {conf.setNumWorkers (1); StormSubmitter.submitTopology (args [0], conf, builder.createTopology ());} else {LocalCluster cluster = new LocalCluster (); cluster.submitTopology ("myTopo", conf, builder.createTopology ());}

2. Spout defines streamId. The accepted bolt needs to define the corresponding definition of componentId and streamId ("streamId_B", "streamId_C") defined in spout to receive. If the id in new Fields ("id") of fiedsGrouping corresponds to the "id" in new Fields ("id", "message") in the corresponding bolt of componentId, it will be grouped with "id".

Package com.cmsz.storm.trading.test;import java.util.Map;import java.util.Random;import backtype.storm.spout.SpoutOutputCollector;import backtype.storm.task.TopologyContext;import backtype.storm.topology.OutputFieldsDeclarer;import backtype.storm.topology.base.BaseRichSpout;import backtype.storm.tuple.Fields;import backtype.storm.tuple.Values;import backtype.storm.utils.Utils;public class ASpout extends BaseRichSpout {SpoutOutputCollector collector; @ Override public void open (Map conf, TopologyContext context, SpoutOutputCollector collector) {this.collector = collector } @ Override public void nextTuple () {Utils.sleep (10); final String [] words = new String [] {"B_nathan", "C_mike", "B_jackson", "C_golda", "B_bertels"}; final Random rand = new Random (); final String word = word [Rand.nextInt (words.length)] If (word.indexOf ("B _") >-1) {collector.emit ("streamId_B", new Values (word));} else if (word.indexOf ("C") >-1) {collector.emit (" streamId_C ", new Values (word));} @ Override public void declareOutputFields (OutputFieldsDeclarer declarer) {declarer.declareStream (" streamId_B ", new Fields (" streamId_B ")) Declarer.declareStream ("streamId_C", new Fields ("streamId_C"));} @ Override public void ack (Object msgId) {super.ack (msgId);} @ Override public void fail (Object msgId) {super.fail (msgId);}} package com.cmsz.storm.trading.test;import java.util.Map;import backtype.storm.task.TopologyContext;import backtype.storm.topology.BasicOutputCollector;import backtype.storm.topology.IBasicBolt;import backtype.storm.topology.OutputFieldsDeclarer Import backtype.storm.tuple.Fields;import backtype.storm.tuple.Tuple;import backtype.storm.tuple.Values;public class BBolt implements IBasicBolt {@ Override public void declareOutputFields (OutputFieldsDeclarer declarer) {declarer.declare (new Fields ("id", "message"));} @ Override public Map getComponentConfiguration () {return null;} @ Override public void prepare (Map stormConf, TopologyContext context) {} @ Override public void execute (Tuple input, BasicOutputCollector collector) {String msg = input.getString (0) System.out.println (msg); collector.emit (new Values (msg,msg+ "BBolt"));} @ Override public void cleanup () {}} package com.cmsz.storm.trading.test;import java.util.Map;import backtype.storm.task.TopologyContext;import backtype.storm.topology.BasicOutputCollector;import backtype.storm.topology.IBasicBolt;import backtype.storm.topology.OutputFieldsDeclarer;import backtype.storm.tuple.Fields;import backtype.storm.tuple.Tuple;import backtype.storm.tuple.Values Public class CBolt implements IBasicBolt {@ Override public void declareOutputFields (OutputFieldsDeclarer declarer) {declarer.declare (new Fields ("id", "message"));} @ Override public Map getComponentConfiguration () {return null;} @ Override public void prepare (Map stormConf, TopologyContext context) {} @ Override public void execute (Tuple input, BasicOutputCollector collector) {String msg = input.getString (0); System.out.println (msg); collector.emit (new Values (msg,msg+ "CBolt")) } @ Override public void cleanup () {}} package com.cmsz.storm.trading.test;import java.util.Map;import backtype.storm.task.TopologyContext;import backtype.storm.topology.BasicOutputCollector;import backtype.storm.topology.IBasicBolt;import backtype.storm.topology.OutputFieldsDeclarer;import backtype.storm.tuple.Fields;import backtype.storm.tuple.Tuple;public class DBolt implements IBasicBolt {@ Override public void declareOutputFields (OutputFieldsDeclarer declarer) {declarer.declare (new Fields ("message")) } @ Override public Map getComponentConfiguration () {return null;} @ Override public void prepare (Map stormConf, TopologyContext context) {} @ Override public void execute (Tuple input, BasicOutputCollector collector) {System.out.println ("DBolt" + input.getString (0);} @ Override public void cleanup () {}} package com.cmsz.storm.trading.test;import java.util.Map;import backtype.storm.task.TopologyContext;import backtype.storm.topology.BasicOutputCollector;import backtype.storm.topology.IBasicBolt Import backtype.storm.topology.OutputFieldsDeclarer;import backtype.storm.tuple.Fields;import backtype.storm.tuple.Tuple;public class EBolt implements IBasicBolt {@ Override public void declareOutputFields (OutputFieldsDeclarer declarer) {declarer.declare (new Fields ("message");} @ Override public Map getComponentConfiguration () {return null;} @ Override public void prepare (Map stormConf, TopologyContext context) {} @ Override public void execute (Tuple input, BasicOutputCollector collector) {System.out.println ("EBolt" + input.getString (0)) } @ Override public void cleanup () {}} at this point, the study on "how to define the storm startup class" is over. I hope I can solve your doubts. The collocation of theory and practice can better help you learn, go and try it! If you want to continue to learn more related knowledge, please continue to follow the website, the editor will continue to work hard to bring you more practical articles!

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.