文档章节

Storm概念详解和工作原理,topology、spout、bolt的细节和API讲解之一

泡海椒
 泡海椒
发布于 2016/02/08 08:50
字数 1376
阅读 2240
收藏 1

storm
datasource -->bolt
|               |
|               |
bolt-->     有向无环图bolt
storm与传统数据库区别
传统数据库先存后计算,而storm则是先算后存甚至不存
传统关系数据库很难部署实时计算,只能部署定时任务统计分析窗口数据
关系型数据库注重事务,并发控制,相对storm来说比较简陋
storm【速度】与hadoop【海量数据】,spark【内存计算框架】等流行的大数据方案
核心代码clojure实用程序python,使用java开发拓扑
wordcount逻辑
sentence spout -->split sentence bolt -->word count bolt -->report bolt

[root@localhost target]# cd ~/soft
[root@localhost soft]# ls
maven  Python-2.7.2  storm-0.9.1  storm-starter  zookeeper-3.3.6
[root@localhost soft]# cd zookeeper-3.3.6
[root@localhost zookeeper-3.3.6]# ls
bin          contrib     ivysettings.xml  NOTICE.txt  zookeeper-3.3.6.jar
build.xml    data        ivy.xml          README.txt  zookeeper-3.3.6.jar.asc
CHANGES.txt  dist-maven  lib              recipes     zookeeper-3.3.6.jar.md5
conf         docs        LICENSE.txt      src         zookeeper-3.3.6.jar.sha1
[root@localhost zookeeper-3.3.6]# cd bin
[root@localhost bin]# ls
README.txt    zkCli.cmd  zkEnv.cmd  zkServer.cmd  zookeeper.out
zkCleanup.sh  zkCli.sh   zkEnv.sh   zkServer.sh
[root@localhost bin]# ./zkCli.sh
[zk: localhost:2181(CONNECTED) 0] ls /
[storm, zookeeper]
[zk: localhost:2181(CONNECTED) 1]
storm管理命令
storm rebalance增加节点之后/activate/deactivate/kill 拓扑名
topology运行流程
storm提交后,会把代码首先存放到nimbus节点的inbox目录下,之后会把当前storm
运行配置生成一个stormconf.ser文件放到nimbus节点的stormdist目录下,在此目录下还有序列化后的toplogy代码文件
2.在设定topology所关联的spout和bolts时,可以同时设置当前spout和bolt的
executor数目和task数目,默认情况下,一个topology的task总和是executor的总和一致的,之后,系统根据workerd的数
尽量平均这些task的执行,work在哪个supervisor节点运行是由本身决定的
3.任务分配好后,niimbus节点会将任务信息提交到zoopeeker集群,同时在zoopecker集群中会有workerbeats
节点,这里存储了所有worker进程的心跳信息
4supervisor节点会不断的轮训zookeeper集群,在zookeeper的assignment节点保存了所有toplogy的任务分配信息
代码存储目录之间的关联关系,supervisor通过轮训此节点的内容,来领取自己的任务,启动worker进程
5.一个topogy运行之后,就会不断的通过spout来发送spout流,通过bolts来不断处理接收的stream流,stream流式误解的
本地运行的提交方式
LocalCluster cluster=new LocalCluster();
cluster.submitTopology(TOPLOGY_NAME,conf,builder.createTopology())
Thread.sleep(2000)
cluster.shutdown();
分布式提交方式:
StormSubmitter。submitToplogy(TOPLOGY_NAME,conf,builder.createTopology())
topology的运行
需要注意的是,在storm代码编写完成之后,需要打成jar包放在nimbus中运行
打包的时候不要加依赖的包,否者会出现重复的配置文件,因为他运行之前会加载
本地的storm。yaml配置文件
storm jar StormTology.jar mainclass
storm守护进程
nimbus
toUI
DRPC
JAR storm jar topology_jar topology_class args
jar是用于提交集群拓扑他运行topology_class main方法,上传jar到nimbus,由nimbus发布到集群
一旦提交,storm会激活拓扑并开始处理topology_class main方法,main方法会调用stormsubmit.submittopology
方法, 并且提供一个唯一的拓扑名, 若这个名字存在那么失败,常见方法是用命令方法来指定拓扑名称
maven新建一个项目
mvn archetype:create/generate  -DgroupId=storm.test -DartifactId=teststorm -DpackageName=cn.dataguru.storm


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>storm.test</groupId>
  <artifactId>teststorm</artifactId>
  <version>0.0.1-SNAPSHOT</version>
  <packaging>jar</packaging>

  <name>teststorm</name>
  <url>http://maven.apache.org</url>

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

  <dependencies>
    <dependency>
      <groupId>junit</groupId>
      <artifactId>junit</artifactId>
      <version>3.8.1</version>
      <scope>test</scope>
    </dependency>
    <dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>3.8.1</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>storm</groupId>
<artifactId>storm</artifactId>
<version>0.9.0.1</version>
<scope>provided</scope>
</dependency>
  </dependencies>
 
  <repositories>
<repository>
<id>clojars.org</id>
<url>http://clojars.org/repo</url>
</repository>
</repositories>


<build>
<plugins>
<plugin>
<artifactId>maven-assembly-plugin</artifactId>
<version>2.2.1</version>
<configuration>
<descriptorRefs>
<descriptorRef>jar-with-dependencies
</descriptorRef>
</descriptorRefs>
<archive>
<manifest>
<mainClass />
</manifest>
</archive>
</configuration>
<executions>
<execution>
<id>make-assembly</id>
<phase>package</phase>
<goals>
<goal>single</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
 
</project>


package cn.dataguru.storm;

import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Tuple;

public class LearningStormBolt extends BaseBasicBolt {
    private static final long serialVersionUID = 1L;

    public void execute(Tuple input, BasicOutputCollector collector) {
        // fetched the field "site" from input tuple.
        String test = input.getStringByField("site");
        // print the value of field "site" on console.
        System.out.println("Name of input site is : " + test);
    }

    public void declareOutputFields(OutputFieldsDeclarer delarer) {

    }
}

package cn.dataguru.storm;

import java.util.HashMap;
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;

public class LearningStormSpout extends BaseRichSpout {
    private static final long serialVersionUID = 1L;
    private SpoutOutputCollector spoutOutputCollector;
    private static final Map<Integer, String> map = new HashMap<Integer, String>();
    static {
        map.put(0, "google");
        map.put(1, "facebook");
        map.put(2, "twitter");
        map.put(3, "youtube");
        map.put(4, "linkedin");
    }

    public void open(Map conf, TopologyContext context,
            SpoutOutputCollector spoutOutputCollector) {
        // Open the spout

        this.spoutOutputCollector = spoutOutputCollector;
    }

    public void nextTuple() {
        // Storm cluster repeatedly calls this method to emit a continuous
        // stream of tuples.
        final Random rand = new Random();
        // generate the random number from 0 to 4.
        int randomNumber = rand.nextInt(5);
        spoutOutputCollector.emit(new Values(map.get(randomNumber)));
    }

    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("site"));
    }
}

package cn.dataguru.storm;

import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.topology.TopologyBuilder;

public class LearningStormTopology {
    public static void main(String[] args) throws AlreadyAliveException,
            InvalidTopologyException {
        // create an instance of TopologyBuilder class
        TopologyBuilder builder = new TopologyBuilder();
        // set the spout class
        builder.setSpout("LearningStormSpout", new LearningStormSpout(), 2);
        // set the bolt class
        builder.setBolt("LearningStormBolt", new LearningStormBolt(), 4)
                .shuffleGrouping("LearningStormSpout");
        Config conf = new Config();
        conf.setDebug(true);
        // create an instance of LocalCluster class for
        // executing topology in local mode.
        LocalCluster cluster = new LocalCluster();
        // LearningStormTopolgy is the name of submitted topology.
        cluster.submitTopology("LearningStormToplogy", conf,
                builder.createTopology());
        try {
            Thread.sleep(10000);
        } catch (Exception exception) {
            System.out.println("Thread interrupted exception : " + exception);
        }
        // kill the LearningStormTopology
        cluster.killTopology("LearningStormToplogy");
        // shutdown the storm test cluster
        cluster.shutdown();
    }
}

mvn install
[root@localhost teststorm]# cd target
[root@localhost target]# ls
archive-tmp       test-classes
classes           teststorm-0.0.1-SNAPSHOT.jar
maven-archiver    teststorm-0.0.1-SNAPSHOT-jar-with-dependencies.jar
maven-status      teststorm-1.0-SNAPSHOT.jar
surefire-reports  teststorm-1.0-SNAPSHOT-jar-with-dependencies.jar

[root@localhost teststorm]# mvn compile exec:java -Dexec:java -Dexec.classpathScope=compile -Dexec.mainClass=cn.dataguru.storm.LearningStormTopology
                                                                                                             cn.dataguru.storm.LearningStormTopology

[root@localhost teststorm]# storm jar teststorm-0.0.1-SNAPSHOT-jar-with-dependencies.jar cn.dataguru.storm.LearningStormTopology

集群方式
conf.setNumWorkers(3);
StormSubmitter.submitTopology("name", conf, builder.createTopology());
以下的注释掉
        cluster.submitTopology("LearningStormToplogy", conf,
                builder.createTopology());
        try {
            Thread.sleep(10000);
            StormSubmitter.submitTopology("name", conf, builder.createTopology());
        } catch (Exception exception) {
            System.out.println("Thread interrupted exception : " + exception);
        }
        // kill the LearningStormTopology
        cluster.killTopology("LearningStormToplogy");
        // shutdown the storm test cluster
        cluster.shutdown();







© 著作权归作者所有

共有 人打赏支持
泡海椒
粉丝 10
博文 279
码字总数 291526
作品 0
成都
程序员
Storm概念讲解和工作原理介绍

Strom的结构 Storm与传统关系型数据库 传统关系型数据库是先存后计算,而storm则是先算后存,甚至不存 传统关系型数据库很难部署实时计算,只能部署定时任务统计分析窗口数据 关系型数据库重...

张超
2015/04/26
0
0
storm 原理简介及单机版安装指南

本文翻译自: https://github.com/nathanmarz/storm/wiki/Tutorial Storm是一个分布式的、高容错的实时计算系统。 Storm对于实时计算的的意义相当于Hadoop对于批处理的意义。Hadoop为我们提供...

大数据之路
2013/07/28
0
5
一篇文带你快速起步Apache Storm

作者介绍 杜亦舒,创业中,技术合伙人,喜欢研究分享技术。个人订阅号:性能与架构。 本文介绍了 Apache Storm 的基本原理和开发方法,包括一个 PDF 和 2 个示例的源码。 内容大纲: Storm是...

杜亦舒
2017/07/11
0
0
storm 原理简介及单机版安装指南(转)

本文翻译自: https://github.com/nathanmarz/storm/wiki/Tutorial Storm是一个分布式的、高容错的实时计算系统。 Storm对于实时计算的的意义相当于Hadoop对于批处理的意义。Hadoop为我们提供...

Jacos
2014/12/03
0
0
六、Storm的高级原语之Transactional Topology

1、什么是Transactional Topology? ○ 是一个每个tuple仅被处理一次的框架 ○ 由Storm0.7引入,于Storm0.9被弃用,被triden取而代之 ○ 底层依靠spoutbolttopologystream抽象的一个特性 2、...

datapro
2015/06/24
0
0

没有更多内容

加载失败,请刷新页面

加载更多

deepin系统使用deepin-wine安装exe程序

deepin自带原生deepin-wine使用命令如下: deepin-wine QQBrowser.exedeepin-wine QQMusicSetup.exe 默认安装的快捷方式位置: /root/.wine/drive_c/'Program Files'/Tencent/QQBrowser/......

临江仙卜算子
46分钟前
2
0
快速get到学习Linux操作系统的点

快速get到学习Linux操作系统的点 Linux是一套免费使用和自由传播的类Unix操作系统,是一个基于POSIX和UNIX的多用户、多任务、支持多线程和多CPU的操作系统。Linux能够运行主要的UNIX工具软件...

linuxCool
53分钟前
3
0
聊聊:Linux分区的那些方案

安装linux的整体步骤其实比较简单,唯一可能值得说明的地方,大概就是linux的分区了。 下面来给大家推荐一些分区方案。 1 分两个区 实际上,很多时候我们只需要分两个区:`/`和交换分区,日常...

Linux就该这么学
今天
1
0
适配器模式和外观模式

适配器模式: 将一个类的接口,转换成客户期望的另一个接口。适配器让原本不兼容的类可以合作无间。 例子: //将Enumeration转换成Iteratorpublic class EnumerationIterator implements Iter...

王怀楼
今天
4
0
7-CXF与Spring整合发布webservice

Spring+CXF整合来管理webservice 实现步骤: 1. 添加cxf.jar 包(集成了Spring.jar、servlet.jar ),spring.jar包 ,servlet.jar 包 2. 编写业务类,通过CXF来发布webservice 员工管理: 方法...

江戸川
今天
4
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部