storm集群配置以及java编写拓扑例子

storm集群配置

storm集群配置

storm配置相当简单

安装

tar -zxvf apache-storm-1.2.2.tar.gz

rm apache-storm-1.2.2.tar.gz

mv apache-storm-1.2.2 storm

sudo vim /etc/profile

    export STORM_HOME=/usr/local/storm

    export PATH=$PATH:$STORM_HOME/bin

source /etc/profile

apt install python

准备 master worker1 worker2 worker3 这四台机器

首先确保你的zookeeper集群能够正常运行worker1 worker2 worker3为zk集群

具体配置参照我的博客https://www.cnblogs.com/ye-hcj/p/9889585.html

修改配置文件

storm.yaml

sudo vim storm.yaml

在四台机器中都加入如下配置

    storm.zookeeper.servers:

    - "worker1"

    - "worker2"

    - "worker3"

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

    supervisor.slots.ports:

    - 6700

    - 6701

    - 6702

    - 6703

    nimbus.seeds: ["master"]

    storm.zookeeper.port: 2181

    // 不加下面这几个你的拓扑直接跑不起来

    nimbus.childopts: "-Xmx1024m" 

    supervisor.childopts: "-Xmx1024m"

    worker.childopts: "-Xmx768m"

启动

在master中运行

    storm nimbus >> /dev/null &

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

在worker1，worker2，worker3中运行

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

    storm logviewer >/dev/null 2>&1 &

直接访问http://master:8080即可

使用java编写拓扑

四个文件如图

pom.xml

    <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">

        <modelVersion>4.0.0</modelVersion>

        <groupId>test</groupId>

        <artifactId>test</artifactId>

        <version>1.0.0</version>

        <name>test</name>

        <description>Test project for spring boot mybatis</description>

        <packaging>jar</packaging>

        <properties>

            <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>

            <maven.compiler.encoding>UTF-8</maven.compiler.encoding>

            <java.version>1.8</java.version>

            <maven.compiler.source>1.8</maven.compiler.source>

            <maven.compiler.target>1.8</maven.compiler.target>

        </properties> 

        <dependencies>

            <dependency>

                <groupId>org.apache.storm</groupId>

                <artifactId>storm-core</artifactId>

                <version>1.2.2</version>

                <scope>provided</scope>

            </dependency>

            <dependency>

                <groupId>junit</groupId>

                <artifactId>junit</artifactId>

                <version>3.8.1</version>

            </dependency>

        </dependencies>

        <build>

            <plugins>

                <plugin>

                    <artifactId>maven-assembly-plugin</artifactId>

                    <configuration>

                        <descriptorRefs>

                            <descriptorRef>jar-with-dependencies</descriptorRef>

                        </descriptorRefs>

                    </configuration>

                    <executions>

                        <execution>

                            <id>make-assembly</id>

                            <phase>package</phase>

                            <goals>

                                <goal>single</goal>

                            </goals>

                        </execution>

                    </executions>

                </plugin>

            </plugins>

        </build>

    </project>

App.java

    package test;

    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 App

    {

        public static void main( String[] args ) throws Exception {

            TopologyBuilder topologyBuilder = new TopologyBuilder();

            topologyBuilder.setSpout("word",new WordSpout(),1);

            topologyBuilder.setBolt("receive",new RecieveBolt(),1).shuffleGrouping("word");

            topologyBuilder.setBolt("print",new ConsumeBolt(),1).shuffleGrouping("receive");

            // 集群运行

            Config config = new Config();

            config.setNumWorkers(3);

            config.setDebug(true);

            StormSubmitter.submitTopology("teststorm", config, topologyBuilder.createTopology());

            // 本地测试

            // Config config = new Config();

            // config.setNumWorkers(3);

            // config.setDebug(true);

            // config.setMaxTaskParallelism(20);

            // LocalCluster cluster = new LocalCluster();

            // cluster.submitTopology("wordCountTopology", config, topologyBuilder.createTopology());

            // Utils.sleep(60000);

            // 执行完毕，关闭cluster

            // cluster.shutdown();

        }

    }

WordSpout.java

    package test;

    import java.util.Map;

    import java.util.Random;

    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 org.apache.storm.utils.Utils;

    public class WordSpout extends BaseRichSpout {

        private static final long serialVersionUID = 6102239192526611945L;

        private SpoutOutputCollector collector;

        Random random = new Random();

        // 初始化tuple的collector

        public void open(Map conf, TopologyContext topologyContext, SpoutOutputCollector collector) {

            this.collector = collector;

        }

        public void nextTuple() {

            // 模拟产生消息队列

            String[] words = {"iphone","xiaomi","mate","sony","sumsung","moto","meizu"};

            final String word = words[random.nextInt(words.length)];

            // 提交一个tuple给默认的输出流

            this.collector.emit(new Values(word));

            Utils.sleep(5000);

        }

        // 声明发送消息的字段名

        public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

            outputFieldsDeclarer.declare(new Fields("word"));

        }

    }

RecieveBolt.java

    package test;

    import java.util.Map;

    import org.apache.storm.task.OutputCollector;

    import org.apache.storm.task.TopologyContext;

    import org.apache.storm.topology.OutputFieldsDeclarer;

    import org.apache.storm.topology.base.BaseRichBolt;

    import org.apache.storm.tuple.Fields;

    import org.apache.storm.tuple.Tuple;

    import org.apache.storm.tuple.Values;

    public class RecieveBolt extends BaseRichBolt {

        private static final long serialVersionUID = -4758047349803579486L;

        private OutputCollector collector;

        public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {

            this.collector = collector;

        }

        public void execute(Tuple tuple) {

            // 将spout传递过来的tuple值进行转换

            this.collector.emit(new Values(tuple.getStringByField("word") + "!!!"));

        }

        // 声明发送消息的字段名

        public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

            outputFieldsDeclarer.declare(new Fields("word"));

        }

    }

ConsumeBolt.java

    package test;

    import java.io.FileWriter;

    import java.io.IOException;

    import java.util.Map;

    import java.util.UUID;

    import org.apache.storm.task.OutputCollector;

    import org.apache.storm.task.TopologyContext;

    import org.apache.storm.topology.OutputFieldsDeclarer;

    import org.apache.storm.topology.base.BaseRichBolt;

    import org.apache.storm.tuple.Tuple;

    public class ConsumeBolt extends BaseRichBolt {

        private static final long serialVersionUID = -7114915627898482737L;

        private FileWriter fileWriter = null;

        private OutputCollector collector;

        public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {

            this.collector = collector;

            try {

                fileWriter = new FileWriter("/usr/local/tmpdata/" + UUID.randomUUID());

                // fileWriter = new FileWriter("C:\\Users\\26401\\Desktop\\test\\" + UUID.randomUUID());

            } catch (IOException e) {

                throw new RuntimeException(e);

            }

        }

        public void execute(Tuple tuple) {

            try {

                String word = tuple.getStringByField("word") + "......." + "\n";

                fileWriter.write(word);

                fileWriter.flush();

                System.out.println(word);

            } catch (IOException e) {

                throw new RuntimeException(e);

            }

        }

        public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

        }

    }

在集群中运行

    storm jar test-1.0.0-jar-with-dependencies.jar test.App // 启动集群

    storm kill teststorm // 结束集群

巴特西

storm集群配置以及java编写拓扑例子

storm集群配置

安装

修改配置文件

使用java编写拓扑

最新文章

热门文章