【预研】Storm Ubuntu 12.04 64 bit 的单机版本的安装和运行

布满荆棘的人生 2022-07-14 00:29 117阅读 0赞

> 基础知识思考整理  
> [http://blog.csdn.net/aganlengzi/article/details/53261391][http_blog.csdn.net_aganlengzi_article_details_53261391]

主要是Storm流计算的简单的环境的搭建和简单的demo的运行。  
当前系统和系统中已经预装的软件：  
其中软件包括jdk1.7.0\_79、python2.7.6、zookeeper3.3.5  
![这里写图片描述][20161121171057829]  
如果没装的话，可以大致按照\[1\]这个安装这些软件。

### 安装ZeroMQ和jzmq ###

Storm底层是基于ZeroMQ消息队列的，所以还需要安装ZeroMQ和其java版本，可以参照\[6\]进行。但是在编译的过程中可能会遇到一些问题，主要是配置和编译的问题，可以通过\[2\]中的方式解决，同时按照\[3\]来进行相关配置。

### 安装Storm ###

参照\[1\]\[3\]可以完成安装，主要是[下载合适版本][Link 1]和相应的文件的配置。

### 测试安装 ###

以上完成整个单机版本Storm的环境的配置，下面是Storm自带的examples中的storm-starter例子（也就是WordCountTopology）的运行。  
启动Storm  
cd /path/to/zookeeper/bin/  
./zkServer.sh start \#启动zookeeper  
cd /path/to/storm/bin  
./storm nimbus &  
./storm supervisor &  
./storm ui &  
启动完成后利用jps查看后台进程：  
![这里写图片描述][20161121171113017]

### 运行demo ###

确保可以成功启动后来进入storm-starter进行WordCountTopology例子的运行：  
实际下面的过程就是strom-starter中README.markdown中的说明和步骤：  
前面熟悉平台和相应例子代码的过程先略过，来看利用Maven运行storm-starter：  
安装Maven3.x  
对Storm首先进行一个本地构建

# Must be run from the top-level directory of the Storm code repository
    $ mvn clean install -DskipTests=true

上面的命令在本地构建Storm并将jar包安装到$HOME/.m2/repository中，后面运行maven命令的时候，Maven能够找到相应的Storm版本等信息。  
将storm-starter打包，便于提交和运行

$ mvn package

生成的包存于storm-starter的target目录下，命名为storm-starter-版本.jar  
后面进行提交的时候可以使用这个包进行提交。

# Example 1: Run the ExclamationTopology in local mode (LocalCluster)
    $ storm jar target/storm-starter-*.jar org.apache.storm.starter.ExclamationTopology
    
    # Example 2: Run the RollingTopWords in remote/cluster mode,
    #            under the name "production-topology"
    $ storm jar storm-starter-*.jar org.apache.storm.starter.RollingTopWords production-topology remote

运行wordCountTopology例子：

cd /path/to/storm/bin
    ./storm jar path/to/generated.jar org.apache.storm.starter.WordCountTopology

运行结果，输出比较多，整理其中一个例子查看：  
![这里写图片描述][20161122115350258]  
可以看到这是“I am at two with nature”的一个统计词频的结果。

源码和相应的注释如下：

/**
     * Licensed to the Apache Software Foundation (ASF) under one
     * or more contributor license agreements.  See the NOTICE file
     * distributed with this work for additional information
     * regarding copyright ownership.  The ASF licenses this file
     * to you under the Apache License, Version 2.0 (the
     * "License"); you may not use this file except in compliance
     * with the License.  You may obtain a copy of the License at
     *
     * http://www.apache.org/licenses/LICENSE-2.0
     *
     * Unless required by applicable law or agreed to in writing, software
     * distributed under the License is distributed on an "AS IS" BASIS,
     * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
     * See the License for the specific language governing permissions and
     * limitations under the License.
     */
    package org.apache.storm.starter;
    
    import org.apache.storm.Config;
    import org.apache.storm.LocalCluster;
    import org.apache.storm.StormSubmitter;
    import org.apache.storm.task.ShellBolt;
    import org.apache.storm.topology.BasicOutputCollector;
    import org.apache.storm.topology.IRichBolt;
    import org.apache.storm.topology.OutputFieldsDeclarer;
    import org.apache.storm.topology.TopologyBuilder;
    import org.apache.storm.topology.base.BaseBasicBolt;
    import org.apache.storm.tuple.Fields;
    import org.apache.storm.tuple.Tuple;
    import org.apache.storm.tuple.Values;
    import org.apache.storm.starter.spout.RandomSentenceSpout;
    
    import java.util.HashMap;
    import java.util.Map;
    
    /**
     * This topology demonstrates Storm's stream groupings and multilang capabilities.
     */
    public class WordCountTopology {
        
        //分词 bolt
        //基于shellBolt实现的SplitSentence类 方便使用多语言 如 python 等
      public static class SplitSentence extends ShellBolt implements IRichBolt {
        
    
        //用pyton写的一个Bolt 构造函数 调用其它语言写的bolt
        public SplitSentence() {
          super("python", "splitsentence.py");
        }
    
        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
          declarer.declare(new Fields("word"));
        }
    
        @Override
        public Map<String, Object> getComponentConfiguration() {
          return null;
        }
      }
        //词频统计 bolt
        //用java实现,所以基类用BaseBasicBolt就够了
      public static class WordCount extends BaseBasicBolt {
        
    
        Map<String, Integer> counts = new HashMap<String, Integer>();
    
        @Override
        public void execute(Tuple tuple, BasicOutputCollector collector) {
          String word = tuple.getString(0);
          Integer count = counts.get(word);
          if (count == null)
            count = 0;
          count++;
          counts.put(word, count);
          //发射单词和计数值
          collector.emit(new Values(word, count));
        }
    
        @Override
        public void declareOutputFields(OutputFieldsDeclarer declarer) {
          declarer.declare(new Fields("word", "count"));
        }
      }
    
      //启动main函数
      //
      public static void main(String[] args) throws Exception {
        //创建一个topology
        TopologyBuilder builder = new TopologyBuilder();
        //创建spout  RandomSentenceSpout是一个随机句子产生器
        builder.setSpout("spout", new RandomSentenceSpout(), 5);
        //创建split  bolt
        builder.setBolt("split", new SplitSentence(), 8).shuffleGrouping("spout");
        //创建count  bolt
        builder.setBolt("count", new WordCount(), 12).fieldsGrouping("split", new Fields("word"));
    
    
        //提交运行
        Config conf = new Config();
        conf.setDebug(true);
    
        if (args != null && args.length > 0) {
          conf.setNumWorkers(3);
          StormSubmitter.submitTopologyWithProgressBar(args[0], conf, builder.createTopology());
        }
        else {
          //最大并行度
          conf.setMaxTaskParallelism(3);
          //提交
          LocalCluster cluster = new LocalCluster();
          cluster.submitTopology("word-count", conf, builder.createTopology());
          Thread.sleep(10000);
          cluster.shutdown();
        }
      }
    }

其中调用的python文件如下：

# Licensed to the Apache Software Foundation (ASF) under one
    # or more contributor license agreements.  See the NOTICE file
    # distributed with this work for additional information
    # regarding copyright ownership.  The ASF licenses this file
    # to you under the Apache License, Version 2.0 (the
    # "License"); you may not use this file except in compliance
    # with the License.  You may obtain a copy of the License at
    #
    # http://www.apache.org/licenses/LICENSE-2.0
    #
    # Unless required by applicable law or agreed to in writing, software
    # distributed under the License is distributed on an "AS IS" BASIS,
    # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    # See the License for the specific language governing permissions and
    # limitations under the License.
    import storm
    
    class SplitSentenceBolt(storm.BasicBolt):
        def process(self, tup):
            words = tup.values[0].split(" ")
            for word in words:
              storm.emit([word])
    
    SplitSentenceBolt().run()

### 参考 ###

\[1\] [http://www.cnblogs.com/stone\_w/p/4487897.html][http_www.cnblogs.com_stone_w_p_4487897.html]  
\[2\] [https://my.oschina.net/mingdongcheng/blog/43009][https_my.oschina.net_mingdongcheng_blog_43009]  
\[3\] [http://blog.csdn.net/zz430581/article/details/37505707][http_blog.csdn.net_zz430581_article_details_37505707]  
\[4\] [http://blog.csdn.net/nb\_vol\_1/article/details/46287625][http_blog.csdn.net_nb_vol_1_article_details_46287625]  
\[5\] [http://blog.csdn.net/nb\_vol\_1/article/details/46287625][http_blog.csdn.net_nb_vol_1_article_details_46287625]  
\[6\] [http://www.cnblogs.com/literoad/archive/2013/03/15/2961035.html][http_www.cnblogs.com_literoad_archive_2013_03_15_2961035.html]

[http_blog.csdn.net_aganlengzi_article_details_53261391]: http://blog.csdn.net/aganlengzi/article/details/53261391
[20161121171057829]: /images/20220714/aee0695e724140f391389a893489d476.png
[Link 1]: http://www.apache.org/dyn/closer.lua/storm/apache-storm-1.0.2/apache-storm-1.0.2.tar.gz
[20161121171113017]: /images/20220714/48a7b31e20504ec8b7e81ad63bf15402.png
[20161122115350258]: /images/20220714/86728bf27b3e4a0bbc982a45415183fa.png
[http_www.cnblogs.com_stone_w_p_4487897.html]: http://www.cnblogs.com/stone_w/p/4487897.html
[https_my.oschina.net_mingdongcheng_blog_43009]: https://my.oschina.net/mingdongcheng/blog/43009
[http_blog.csdn.net_zz430581_article_details_37505707]: http://blog.csdn.net/zz430581/article/details/37505707
[http_blog.csdn.net_nb_vol_1_article_details_46287625]: http://blog.csdn.net/nb_vol_1/article/details/46287625
[http_www.cnblogs.com_literoad_archive_2013_03_15_2961035.html]: http://www.cnblogs.com/literoad/archive/2013/03/15/2961035.html