Storm编程入门API系列之Storm的Topology多个Workers数目控制实现-低调大师

Storm编程入门API系列之Storm的Topology多个Workers数目控制实现

2017-11-20 538

　　继续编写

StormTopologyMoreWorker.java

package zhouls.bigdata.stormDemo;

import java.util.Map;

import org.apache.storm.Config;
import org.apache.storm.LocalCluster;
import org.apache.storm.StormSubmitter;
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.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;
import org.apache.storm.utils.Utils;


public class StormTopologyMoreWorker {
    
    public static class MySpout extends BaseRichSpout{
        private Map conf;
        private TopologyContext context;
        private SpoutOutputCollector collector;
        public void open(Map conf, TopologyContext context,
                SpoutOutputCollector collector) {
            this.conf = conf;
            this.collector = collector;
            this.context = context;
        }

        int num = 0; 
        public void nextTuple() {
            num++;
            System.out.println("spout:"+num);
            this.collector.emit(new Values(num));
            Utils.sleep(1000);
        }

        public void declareOutputFields(OutputFieldsDeclarer declarer) {
            declarer.declare(new Fields("num"));
        }
        
    }
    
    
    
    public static class MyBolt extends BaseRichBolt{
        
        private Map stormConf;
        private TopologyContext context;
        private OutputCollector collector;
        public void prepare(Map stormConf, TopologyContext context,
                OutputCollector collector) {
            this.stormConf = stormConf;
            this.context = context;
            this.collector = collector;
        }
        
        int sum = 0;
        public void execute(Tuple input) {
            Integer num = input.getIntegerByField("num");
            sum += num;
            System.out.println("sum="+sum);
        }

        public void declareOutputFields(OutputFieldsDeclarer declarer) {
            
        }
        
    }
    
    
    
    public static void main(String[] args) {
        TopologyBuilder topologyBuilder = new TopologyBuilder();
        String spout_id = MySpout.class.getSimpleName();
        String bolt_id = MyBolt.class.getSimpleName();
        
        topologyBuilder.setSpout(spout_id, new MySpout());
        topologyBuilder.setBolt(bolt_id, new MyBolt()).shuffleGrouping(spout_id);
        
        
        Config config = new Config();
        config.setNumWorkers(2);
        String topology_name = StormTopologyMoreWorker.class.getSimpleName();
        if(args.length==0){
            //在本地运行
            LocalCluster localCluster = new LocalCluster();
            localCluster.submitTopology(topology_name, config, topologyBuilder.createTopology());
        }else{
            //在集群运行
            try {
                StormSubmitter.submitTopology(topology_name, config, topologyBuilder.createTopology());
            } catch (AlreadyAliveException e) {
                e.printStackTrace();
            } catch (InvalidTopologyException e) {
                e.printStackTrace();
            } catch (AuthorizationException e) {
                e.printStackTrace();
            }
        }
        
    }

}

　　打jar包

[hadoop@master jar]$ pwd
/home/hadoop/app/apache-storm-1.0.2/jar
[hadoop@master jar]$ ll
total 8
-rw-r--r-- 1 hadoop hadoop 4869 Jul 27 22:17 StormTopology.jar
[hadoop@master jar]$ rz

[hadoop@master jar]$ ll
total 16
-rw-r--r-- 1 hadoop hadoop 4869 Jul 27 22:17 StormTopology.jar
-rw-r--r-- 1 hadoop hadoop 4992 Jul 27 22:39 StormTopologyMoreWorker.jar

提交作业之前

[hadoop@master apache-storm-1.0.2]$ pwd
/home/hadoop/app/apache-storm-1.0.2
[hadoop@master apache-storm-1.0.2]$ ll
total 208
drwxrwxr-x  2 hadoop hadoop  4096 May 21 17:18 bin
-rw-r--r--  1 hadoop hadoop 82317 Jul 27  2016 CHANGELOG.md
drwxrwxr-x  2 hadoop hadoop  4096 Jul 27 20:12 conf
drwxrwxr-x  3 hadoop hadoop  4096 Jul 27  2016 examples
drwxrwxr-x 17 hadoop hadoop  4096 May 21 17:18 external
drwxrwxr-x  2 hadoop hadoop  4096 Jul 27  2016 extlib
drwxrwxr-x  2 hadoop hadoop  4096 Jul 27  2016 extlib-daemon
drwxrwxr-x  2 hadoop hadoop  4096 Jul 27 22:39 jar
drwxrwxr-x  2 hadoop hadoop  4096 May 21 17:18 lib
-rw-r--r--  1 hadoop hadoop 32101 Jul 27  2016 LICENSE
drwxrwxr-x  2 hadoop hadoop  4096 May 21 17:18 log4j2
drwxrwxr-x  2 hadoop hadoop  4096 May 21 19:05 logs
-rw-r--r--  1 hadoop hadoop   981 Jul 27  2016 NOTICE
drwxrwxr-x  6 hadoop hadoop  4096 May 21 17:18 public
-rw-r--r--  1 hadoop hadoop 15287 Jul 27  2016 README.markdown
-rw-r--r--  1 hadoop hadoop     6 Jul 27  2016 RELEASE
-rw-r--r--  1 hadoop hadoop 23774 Jul 27  2016 SECURITY.md
[hadoop@master apache-storm-1.0.2]$ bin/storm jar jar/StormTopologyMoreWorker.jar zhouls.bigdata.stormDemo.StormTopologyMoreWorker aaa
Running: /home/hadoop/app/jdk/bin/java -client -Ddaemon.name= -Dstorm.options= -Dstorm.home=/home/hadoop/app/apache-storm-1.0.2 -Dstorm.log.dir=/home/hadoop/app/apache-storm-1.0.2/logs -Djava.library.path=/usr/local/lib:/opt/local/lib:/usr/lib -Dstorm.conf.file= -cp /home/hadoop/app/apache-storm-1.0.2/lib/log4j-api-2.1.jar:/home/hadoop/app/apache-storm-1.0.2/lib/kryo-3.0.3.jar:/home/hadoop/app/apache-storm-1.0.2/lib/storm-rename-hack-1.0.2.jar:/home/hadoop/app/apache-storm-1.0.2/lib/log4j-core-2.1.jar:/home/hadoop/app/apache-storm-1.0.2/lib/slf4j-api-1.7.7.jar:/home/hadoop/app/apache-storm-1.0.2/lib/minlog-1.3.0.jar:/home/hadoop/app/apache-storm-1.0.2/lib/objenesis-2.1.jar:/home/hadoop/app/apache-storm-1.0.2/lib/clojure-1.7.0.jar:/home/hadoop/app/apache-storm-1.0.2/lib/servlet-api-2.5.jar:/home/hadoop/app/apache-storm-1.0.2/lib/log4j-slf4j-impl-2.1.jar:/home/hadoop/app/apache-storm-1.0.2/lib/log4j-over-slf4j-1.6.6.jar:/home/hadoop/app/apache-storm-1.0.2/lib/storm-core-1.0.2.jar:/home/hadoop/app/apache-storm-1.0.2/lib/disruptor-3.3.2.jar:/home/hadoop/app/apache-storm-1.0.2/lib/asm-5.0.3.jar:/home/hadoop/app/apache-storm-1.0.2/lib/reflectasm-1.10.1.jar:jar/StormTopologyMoreWorker.jar:/home/hadoop/app/apache-storm-1.0.2/conf:/home/hadoop/app/apache-storm-1.0.2/bin -Dstorm.jar=jar/StormTopologyMoreWorker.jar zhouls.bigdata.stormDemo.StormTopologyMoreWorker aaa
6687 [main] INFO  o.a.s.StormSubmitter - Generated ZooKeeper secret payload for MD5-digest: -5319119555480935017:-9144362215090188990
7204 [main] INFO  o.a.s.s.a.AuthUtils - Got AutoCreds []
7785 [main] INFO  o.a.s.StormSubmitter - Uploading topology jar jar/StormTopologyMoreWorker.jar to assigned location: /home/hadoop/data/storm/nimbus/inbox/stormjar-ea1d6383-ca31-4033-b784-e06
856000894.jar
7875 [main] INFO  o.a.s.StormSubmitter - Successfully uploaded topology jar to assigned location: /home/hadoop/data/storm/nimbus/inbox/stormjar-ea1d6383-ca31-4033-b784-e06856000894.jar
7876 [main] INFO  o.a.s.StormSubmitter - Submitting topology StormTopologyMoreWorker in distributed mode with conf {"storm.zookeeper.topology.auth.scheme":"digest","storm.zookeeper.topology.auth.payload":"-5319119555480935017:-9144362215090188990","topology.workers":2}
8962 [main] INFO  o.a.s.StormSubmitter - Finished submitting topology: StormTopologyMoreWorker
[hadoop@master apache-storm-1.0.2]$

　　提交之后

　　为什么，会是如上的数字呢？大家要学，就要深入去学和理解。

　　因为，我之前运行的StormTopology没有停掉

　　现在呢，我将之前运行的StormTopology给停掉，然后，再来看。非常重要

　　即提示，我们，30秒之后，kill掉。如果大家等不及，可以设置时间短些

　　为什么，会是如上的数字呢？大家要学，就要深入去学和理解。

本文转自大数据躺过的坑博客园博客，原文链接：http://www.cnblogs.com/zlslch/p/7247860.html，如需转载请自行联系原作者

微信关注我们

原文链接：https://yq.aliyun.com/articles/365566

转载内容版权归作者及来源网站所有！

低调大师中文资讯倾力打造互联网数据资讯、行业资源、电子商务、移动互联网、网络营销平台。持续更新报道IT业界、互联网、市场资讯、驱动更新,是最及时权威的产业资讯及硬件资讯报道平台。

Hadoop can't load native lib

前几天部门刚来不久的小同学装Hadoop一直失败，报libhadoop加载错误解决不了.过去查看下相关依赖lib都解决了，JAVA_LIBRARY_PATH也指定正确位置了，为啥还会失败呢？直接强制加载一次nativelib再看发现原来在另外一个目录有人安装了libhadoop，但是版本是错误的，很多依赖都找不到，而HADOOP启动时正加载了这个路径下的libhadoop导致失败的，这是什么原因呢？查看HADOOP在nativelib的加载过程，最后跟踪到classloader的loadLibrary方法发现： 1 2 usr_paths = initializePath( "java.library.path" ); sys_paths = initializePath( "sun.boot.library.path" ); 原来在加载javalib时，classloader是优先加载sys_paths然后才是usr_paths. 我也学习了：）本文转自MIKE老毕 51CTO博客，原文链接：http://blog.51cto.com/boylook/1367254，如需转载请...

2017-11-21

581

在Hive0.10之前hiverc file都是从HIVE_HOME/bin下加载的，而在之后的加载顺序如下： 1.HIVE_HOME/bin为了保证兼容 2.HIVE_CONF_DIR/ 3.USER_HOME/ 另外一个常见的问题就是hiverc的注释，这个容易受到linux注释习惯的影响“#”，其实Hive里所有的注释都是用“--”的，否则会导致hive无法执行=。=这个一定要谨慎！本文转自MIKE老毕 51CTO博客，原文链接：http://blog.51cto.com/boylook/1365741，如需转载请自行联系原作者

2017-11-21

617

资源下载

更多资源

腾讯云软件源

为解决软件依赖安装时官方源访问速度慢的问题，腾讯云为一些软件搭建了缓存服务。您可以通过使用腾讯云软件源站来提升依赖包的安装速度。为了方便用户自由搭建服务架构，目前腾讯云软件源站支持公网访问和内网访问。

Nacos

Nacos /nɑ:kəʊs/ 是 Dynamic Naming and Configuration Service 的首字母简称，一个易于构建 AI Agent 应用的动态服务发现、配置管理和AI智能体管理平台。Nacos 致力于帮助您发现、配置和管理微服务及AI智能体应用。Nacos 提供了一组简单易用的特性集，帮助您快速实现动态服务发现、服务配置、服务元数据、流量管理。Nacos 帮助您更敏捷和容易地构建、交付和管理微服务平台。

Rocky Linux

Rocky Linux（中文名：洛基）是由Gregory Kurtzer于2020年12月发起的企业级Linux发行版，作为CentOS稳定版停止维护后与RHEL（Red Hat Enterprise Linux）完全兼容的开源替代方案，由社区拥有并管理，支持x86_64、aarch64等架构。其通过重新编译RHEL源代码提供长期稳定性，采用模块化包装和SELinux安全架构，默认包含GNOME桌面环境及XFS文件系统，支持十年生命周期更新。

Sublime Text

Sublime Text具有漂亮的用户界面和强大的功能，例如代码缩略图，Python的插件，代码段等。还可自定义键绑定，菜单和工具栏。Sublime Text 的主要功能包括：拼写检查，书签，完整的 Python API ， Goto 功能，即时项目切换，多选择，多窗口等等。Sublime Text 是一个跨平台的编辑器，同时支持Windows、Linux、Mac OS X等操作系统。