Storm累计求和中使用各种分组Grouping-低调大师

Storm累计求和中使用各种分组Grouping

2017-11-18 424

Shuffle Grouping: 随机分组，随机派发stream里面的tuple，保证bolt中的每个任务接收到的tuple数目相同.(它能实现较好的负载均衡)

Fields Grouping：按字段分组，比如按userid来分组，具有同样userid的tuple会被分到同一任务，而不同的userid则会被分配到不同的任务

All Grouping：广播发送,对于每一个tuple,Bolts中的所有任务都会收到.

Global Grouping: 全局分组，这个tuple被分配到storm中的一个bolt的其中一个task.再具体一点就是分配给id值最低的那个task.

Non Grouping: 随机分派，意思是说stream不关心到底谁会收到它的tuple.目前他和Shuffle grouping是一样的效果,

Direct Grouping: 直接分组,这是一种比较特别的分组方法，用这种分组意味着消息的发送者具体由消息接收者的哪个task处理这个消息.只有被声明为Direct Stream的消息流可以声明这种分组方法.而且这种消息tuple必须使用emitDirect方法来发射.消息处理者可以通过TopologyContext来或者处理它的消息的taskid (OutputCollector.emit方法也会返回taskid)

Fields Grouping 的代码

  1 /**
  2  * 数字累加求和
  3  * 先添加storm依赖
  4  */
  5 public class LocalTopologySumFieldsGrouping {
  6     /**
  7      * spout需要继承baserichspout，实现未实现的方法
  8      * @author Administrator
  9      *
 10      */
 11     public static class MySpout extends BaseRichSpout{
 12         private Map conf;
 13         private TopologyContext context;
 14         private SpoutOutputCollector collector;
 15         
 16         /**
 17          * 初始化方法，只会执行一次
 18          * 在这里面可以写一个初始化的代码
 19          * Map conf：其实里面保存的是topology的一些配置信息
 20          * TopologyContext context：topology的上下文，类似于servletcontext
 21          * SpoutOutputCollector collector：发射器，负责向外发射数据(tuple)
 22          */
 23         @Override
 24         public void open(Map conf, TopologyContext context,
 25                 SpoutOutputCollector collector) {
 26             this.conf = conf;
 27             this.context = context;
 28             this.collector = collector;
 29         }
 30 
 31         int num = 1;
 32         /**
 33          * 这个方法是spout中最重要的方法，
 34          * 这个方法会被storm框架循环调用，可以理解为这个方法是在一个while循环之内
 35          * 每调用一次，会向外发射一条数据
 36          */
 37         @Override
 38         public void nextTuple() {
 39             System.out.println("spout发射："+num);
 40             //把数据封装到values中，称为一个tuple，发射出去
 41             this.collector.emit(new Values(num++,num%2));
 42             Utils.sleep(1000);
 43         }
 44         
 45         /**
 46          * 声明输出字段
 47          */
 48         @Override
 49         public void declareOutputFields(OutputFieldsDeclarer declarer) {
 50             //给values中的数据起个名字，方便后面的bolt从这个values中取数据
 51             //fields中定义的参数和values中传递的数值是一一对应的
 52             declarer.declare(new Fields("num","flag"));
 53         }
 54     }
 55     
 56     /**
 57      * 自定义bolt需要实现baserichbolt
 58      * @author Administrator
 59      *
 60      */
 61     public static class MyBolt extends BaseRichBolt{
 62         private Map stormConf; 
 63         private TopologyContext context;
 64         private OutputCollector collector;
 65         
 66         /**
 67          * 和spout中的open方法意义一样
 68          */
 69         @Override
 70         public void prepare(Map stormConf, TopologyContext context,
 71                 OutputCollector collector) {
 72             this.stormConf = stormConf;
 73             this.context = context;
 74             this.collector = collector;
 75         }
 76 
 77         int sum = 0;
 78         /**
 79          * 是bolt中最重要的方法，当spout发射一个tuple出来，execute也会被调用，需要对spout发射出来的tuple进行处理
 80          */
 81         @Override
 82         public void execute(Tuple input) {
 83             //input.getInteger(0);//也可以根据角标获取tuple中的数据
 84             Integer value = input.getIntegerByField("num");
 85             System.out.println("线程id："+Thread.currentThread().getId()+",值："+value);
 86             //sum+=value;
 87             //System.out.println("和："+sum);
 88         }
 89         
 90         /**
 91          * 声明输出字段
 92          */
 93         @Override
 94         public void declareOutputFields(OutputFieldsDeclarer declarer) {
 95             //在这没必要定义了，因为execute方法中没有向外发射tuple，所以就不需要声明了。
 96             //如果nextTuple或者execute方法中向外发射了tuple，那么declareOutputFields必须要声明，否则不需要声明
 97         }
 98         
 99     }
100     /**
101      * 注意：在组装topology的时候，组件的id在定义的时候，名称不能以__开头。__是系统保留的
102      * @param args
103      */
104     public static void main(String[] args) {
105         //组装topology
106         TopologyBuilder topologyBuilder = new TopologyBuilder();
107         topologyBuilder.setSpout("spout1", new MySpout());
108         //.shuffleGrouping("spout1"); 表示让MyBolt接收MySpout发射出来的tuple
109         topologyBuilder.setBolt("bolt1", new MyBolt(),3).fieldsGrouping("spout1", new Fields("flag"));
110         
111         //创建本地storm集群
112         LocalCluster localCluster = new LocalCluster();
113         localCluster.submitTopology("sumTopology", new Config(), topologyBuilder.createTopology());
114     }
115     
116 }

ShuffleGrouping代码

  1 /**
  2  * 数字累加求和
  3  * 先添加storm依赖
  4  */
  5 public class LocalTopologySumShufferGrouping {
  6     /**
  7      * spout需要继承baserichspout，实现未实现的方法
  8      * @author Administrator
  9      *
 10      */
 11     public static class MySpout extends BaseRichSpout{
 12         private Map conf;
 13         private TopologyContext context;
 14         private SpoutOutputCollector collector;
 15         
 16         /**
 17          * 初始化方法，只会执行一次
 18          * 在这里面可以写一个初始化的代码
 19          * Map conf：其实里面保存的是topology的一些配置信息
 20          * TopologyContext context：topology的上下文，类似于servletcontext
 21          * SpoutOutputCollector collector：发射器，负责向外发射数据(tuple)
 22          */
 23         @Override
 24         public void open(Map conf, TopologyContext context,
 25                 SpoutOutputCollector collector) {
 26             this.conf = conf;
 27             this.context = context;
 28             this.collector = collector;
 29         }
 30 
 31         int num = 1;
 32         /**
 33          * 这个方法是spout中最重要的方法，
 34          * 这个方法会被storm框架循环调用，可以理解为这个方法是在一个while循环之内
 35          * 每调用一次，会向外发射一条数据
 36          */
 37         @Override
 38         public void nextTuple() {
 39             System.out.println("spout发射："+num);
 40             //把数据封装到values中，称为一个tuple，发射出去
 41             this.collector.emit(new Values(num++));
 42             Utils.sleep(1000);
 43         }
 44         
 45         /**
 46          * 声明输出字段
 47          */
 48         @Override
 49         public void declareOutputFields(OutputFieldsDeclarer declarer) {
 50             //给values中的数据起个名字，方便后面的bolt从这个values中取数据
 51             //fields中定义的参数和values中传递的数值是一一对应的
 52             declarer.declare(new Fields("num"));
 53         }
 54         
 55     }
 56     
 57     
 58     /**
 59      * 自定义bolt需要实现baserichbolt
 60      * @author Administrator
 61      *
 62      */
 63     public static class MyBolt extends BaseRichBolt{
 64         private Map stormConf; 
 65         private TopologyContext context;
 66         private OutputCollector collector;
 67         
 68         /**
 69          * 和spout中的open方法意义一样
 70          */
 71         @Override
 72         public void prepare(Map stormConf, TopologyContext context,
 73                 OutputCollector collector) {
 74             this.stormConf = stormConf;
 75             this.context = context;
 76             this.collector = collector;
 77         }
 78 
 79         int sum = 0;
 80         /**
 81          * 是bolt中最重要的方法，当spout发射一个tuple出来，execute也会被调用，需要对spout发射出来的tuple进行处理
 82          */
 83         @Override
 84         public void execute(Tuple input) {
 85             //input.getInteger(0);//也可以根据角标获取tuple中的数据
 86             Integer value = input.getIntegerByField("num");
 87             System.out.println("线程id："+Thread.currentThread().getId()+",值："+value);//这样可以知道哪个线程接收到这个数据了.
 88             //sum+=value;
 89             //System.out.println("和："+sum);
 90         }
 91         
 92         /**
 93          * 声明输出字段
 94          */
 95         @Override
 96         public void declareOutputFields(OutputFieldsDeclarer declarer) {
 97             //在这没必要定义了，因为execute方法中没有向外发射tuple，所以就不需要声明了。
 98             //如果nextTuple或者execute方法中向外发射了tuple，那么declareOutputFields必须要声明，否则不需要声明
 99         }
100         
101     }
102     /**
103      * 注意：在组装topology的时候，组件的id在定义的时候，名称不能以__开头。__是系统保留的
104      * @param args
105      */
106     public static void main(String[] args) {
107         //组装topology
108         TopologyBuilder topologyBuilder = new TopologyBuilder();
109         topologyBuilder.setSpout("spout1", new MySpout());
110         //.shuffleGrouping("spout1"); 表示让MyBolt接收MySpout发射出来的tuple
111         topologyBuilder.setBolt("bolt1", new MyBolt(),3).globalGrouping("spout1");
112         
113         //创建本地storm集群
114         LocalCluster localCluster = new LocalCluster();
115         localCluster.submitTopology("sumTopology", new Config(), topologyBuilder.createTopology());
116     }
117 }

本文转自SummerChill博客园博客，原文链接：http://www.cnblogs.com/DreamDrive/p/5793975.html，如需转载请自行联系原作者

微信关注我们

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

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

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

3 weekend110的hadoop中的RPC框架实现机制 + hadoop中的RPC应用实例demo

hadoop中的RPC框架实现机制 RPC是Remotr Process Call，进程间的远程过程调用，不是在一个jvm里。即，Controller拿不到Service的实例对象。 hadoop中的RPC应用实例demo 在windows是调用端，在linux里是服务端。在这里，需要LoginServiceinterface.java 停止出错误了，很明显。这是个很好的思考题？本文转自大数据躺过的坑博客园博客，原文链接：http://www.cnblogs.com/zlslch/p/5894446.html，如需转载请自行联系原作者

2017-11-19

564

我们在写Hadoop--map/reduce程序时，遇到使用按文件url来分析文件----------多表连接的DistributedCache方式，看不懂使用extends Configured implements Tool的方式，就查了一下http://hadoop.apache.org 上面对该Tool接口及其使用做了说明： @InterfaceAudience.Public @InterfaceStability.Stable publicinterfaceTool//Tool接口继承了Configurable extendsConfigurable //Tool接口可以支持处理通用的命令行选项，它是所有Map-Reduce程序的都可用的一个标准接口，下面是一个典型用例： publicclassMyAppextendsConfiguredimplementsTool{ publicintrun(String[]args)throwsException{ //ToolRunner要处理的Configuration，Tool通过ToolRunner调用ToolRunner...

2017-11-19

632

资源下载

更多资源

Mario

马里奥是站在游戏界顶峰的超人气多面角色。马里奥靠吃蘑菇成长，特征是大鼻子、头戴帽子、身穿背带裤，还留着胡子。与他的双胞胎兄弟路易基一起，长年担任任天堂的招牌角色。

Nacos

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

Spring

Spring框架（Spring Framework）是由Rod Johnson于2002年提出的开源Java企业级应用框架，旨在通过使用JavaBean替代传统EJB实现方式降低企业级编程开发的复杂性。该框架基于简单性、可测试性和松耦合性设计理念，提供核心容器、应用上下文、数据访问集成等模块，支持整合Hibernate、Struts等第三方框架，其适用范围不仅限于服务器端开发，绝大多数Java应用均可从中受益。

Rocky Linux

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