分布式storm-1.2.2

　

storm

分布式storm-1.2.2

Storm是一个开源的分布式实时大数据处理系统，可以处理海量数据，具有高可靠，高容错，高扩展的特点，适用于数据实时处理而非批处理.

修改配置文件storm.yaml

#配置zk地址（配置zk的主机名，如果zk集群使用的不是默认端口。那么还需要配置#strom.zookeeper.port）

storm.zookeeper.servers:

     - "master"

     - "slave1"

     - "slave2"

#用于配置storm存出少量文件的路径，在storm启动的时候会自动创建，建议用户提前手动创建

storm.local.dir: "/home/hadoop/storm/stormdata"

#配置storm主控节点的地址，多个可以解决单点故障问题

nimbus.seeds: ["master"]

#指定运行webui的主机（在该主机上键入storm ui &），可以在浏览器上输入#192.168.218.134:8181回车

ui.host: 192.168.218.134

#访问webui时的端口号

ui.port: 8181

#配置工作节点上的进程端口，一个端口对应一个worker，在实际的生产环境中，我们需

#要根据实际的物理配置以及每个节点上的负载情况来配置这个端口的数量

supervisor.slots.ports:

     - 6700

     - 6701

     - 6702

     - 6703

根据storm.yaml中所配置的storm.local.dir创建目录

mkdir -p /home/hadoop/storm/stormdata

同步到其他节点

scp -r /home/hadoop/storm hadoop@slave1:/home/hadoop

scp -r /home/hadoop/storm hadoop@slave2:/home/hadoop

配置集群所有节点storm环境变量

vi ~/.bash_profile

source ~/.bash_profile

export STORM_HOME=/home/hadoop/storm/apache-storm-1.2.2

${STORM_HOME}/bin

启动zk

启动storm

master节点 192.168.218.133

strom nimbus &

storm ui &

slave1节点 192.168.218.135

storm supervisor &

storm logviewer &

slave2节点 192.168.218.134

storm supervisor &

storm logviewer &

检查storm集群启动是否正常

webui对应的链接:http://192.168.218.133:8181

superviOSr的log功能需要分别在supervisor的主机上启动logviewer进程

storm logviewer &

点击log跳转后显示如下：

http://slave1:8000/daemonlog?file=supervisor.log

显示失败原因：该host对应的主机上没有启动logviewer进程，本地windows无法识别slave1的主机名

解决办法：1将链接中的slave1修改为对应的ip，这里是192.168.218.135

         2 修改windows文件C:\Windows\System32\drivers\etc\hosts（推荐）

192.168.218.134 slave2

192.168.218.135 slave1

storm报错

当在storm的nimbus.log中发现以下异常（这个时候jps任然可以看到nimbus进程）

java.lang.runtimeexception: No nimbus leader participant host found, have you started your nimbus hosts?

可以尝试如下解决办法：删除storm提交给zk的配置信息

rmr /storm

脚本启动和停止Storm集群

以下脚本需要在hostname=master运行

启动storm

vi start-storm-all.sh

chmod u+x start-storm-all.sh

#!/bin/bash

#启动storm集群

#在本机上启动nimbus进程和ui进程

echo "start nimbus and ui in master"

nohup ${STORM_HOME}/bin/storm nimbus >/dev/null 2>&1 &

nohup ${STORM_HOME}/bin/storm ui >/dev/null 2>&1 &



#在指定作为supervisor的服务器上启动supervisor和logviewer进程

#supervisorHosts指定supervisor服务器的hostname

supervisorHosts=('slave1' 'slave2')

for supervisor in ${supervisorHosts[@]}

do

        echo "start supervisor and logviewer in ${supervisor}"

        ssh hadoop@${supervisor} "source ~/.bash_profile && nohup ${STORM_HOME}/bin/storm supervisor >/dev/n

ull 2>&1" >/dev/null 2>&1 &

        ssh hadoop@${supervisor} "source ~/.bash_profile && nohup ${STORM_HOME}/bin/storm logviewer >/dev/nu

ll 2>&1" >/dev/null 2>&1 &

done

停止storm的脚本开发（nimbus ui logviewer supervisor）

vi stop-storm-all.sh

chmod u+x stop-storm-all.sh

#!/bin/bash

#停止storm集群



#停止本机上的nimbus和ui进程

echo "stop nimbus and ui in master"

kill -9 `ps -ef | grep daemon.nimbus | awk '{print $2}' | head -n 1`

kill -9 `ps -ef | grep ui.core | awk '{print $2}' | head -n 1`



#停止supervisor节点上的supervisor和logviewer的进程

supervisorHost=('slave1' 'slave2')

for supervisor in ${supervisorHost[@]}

do

        echo "stop supervisor and logviewer in $supervisor"

        ssh ${supervisor} "kill -9 `ssh ${supervisor} ps -ef | grep daemon.supervisor | awk '{print $2}' | h

ead -n 1`" >/dev/null 2>&1

        ssh ${supervisor} "kill -9 `ssh ${supervisor} ps -ef | grep daemon.logviewer | awk '{print $2}' | he

ad -n 1`" >/dev/null 2>&1

done

提交jar到storm集群运行

storm jar xxx.jar xxx xxx

第一个xxx是storm程序打包的包名

第二个xxx是运行主程序的路径

第三个xxx是输入的参数，可以有多个，空格分开

storm核心概念

master node	主节点，运行nimbus守护进程，负责在集群中分发代码，为工作节点分配任务，并监控故障
worker nodes	从节点，运行supervisor守护进程，负责接收工作任务，根据nimbus指令来启动或停止工作进程
spout	数据流的源头，从外部接收数据，是topology的消息生产者
bolt	逻辑处理单元，Spouts向Bolts发射数据，Bolts接收数据进行逻辑处理后再把结果发射出去
stream	数据流，一个无序的tuple序列
tuple	数据流中的数据单元，包含一个或多个键值对的列表
topology	拓扑，Spouts和Bolts连接起来后形成了Topology，其中定义了整个应用的实时处理逻辑，它是一个有向图，定点是计算单元，边是数据流。始于Spout，Spout向一个或多个Bolt发射数据，Bolt拥有处理逻辑，Bolt的输出可以发射给其它的Bolt作为它们的输入，Storm会保持Topology一直运行，除非kill掉它。
task	task是storm中进行计算的最小运行单位，默认一个executor执行一个task
worker process (工作进程)	负责实际执行task，conf.setNumWorkers(3);//设置topology的worker数量，默认一个topology用一个worker进程。 worker里面包含一个或多个线程executor，一个executor会处理一个或多个task，一个task就是一个节点类的实例对象。每一个worker进程使用一个单独的端口来收取消息。
Nimbus	storm集群主节点，负责资源分配和任务调度，我们提交任务和截止任务都是在nimbus上执行的
Supervisor	storm集群工作节点，接受Nimbus分配任务，管理所有worker，它和nimbus都是无状态的，它和nimbus之间的协作是通过zk集群来完成的
Stream Grouping	流分组，消息分发策略（定义每个bolt接收何种输入），简单理解就是控制Tuple的路由，定义Tuple在Topology中如何流动，每个Spout或Bolt都会在集群中执行多个任务，每一个任务都对应为一个线程的执行。Stream Grouping定义的就是如何从一个Task集合向其它Task集合发送Tuple。
Shuffle Grouping	随机分配，可以让每个Bolt获得数量均衡的Tuple
field Grouping	field名字相同的Tuple会被组织在一起去往同一个Task
Gobal Grouping	全局统一分组，所有数据流都流往一个Bolt，多进行汇总统计
All Grouping	向每一个实例都发送一次，主要用于发送信号
Executor	由worker进程创建的线程 一个executor会执行一个或多个task，默认1个 parallelism_hint
slot	storm.yaml配置文件中通过supervisor.slots.ports设置了端口，每一个端口提供一个slot，即supervisor的数量乘以ports数量就是storm集群中slot总数 一个worker会消耗一个slot，所以solt数量就是集群中总的worker数量 当slot全部分配完，就不能再加载新的topology。

storm之java api

package storm2;



import java.io.BufferedReader;

import java.io.BufferedWriter;

import java.io.fileinputstream;

import java.io.filenotfoundException;

import java.io.Fileoutputstream;

import java.io.IOException;

import java.io.inputstreamreader;

import java.io.outputstreamwriter;

import java.util.Map;



import org.apache.storm.Config;

import org.apache.storm.Localcluster;

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 SumTopology {



/**

 * 向后端发射tuple数据流

 */

public static class SumSpout extends BaseRichSpout {



private static final long serialversionuid = 1L;



private String number;



private SpoutOutputCollector collector;



private BufferedReader reader;



/**

* open()方法中是ISpout接口中定义，在Spout组件初始化时被调用。

* open()接受三个参数:一个包含Storm配置的Map,一个TopologyContext对象，提供了topology中组件的信息,SpoutOutputCollector对象提供发射tuple的方法。

* 在这个例子中,我们不需要执行初始化,只是简单的存储在一个SpoutOutputCollector实例变量。

*/

@SuppressWarnings("rawtypes")

@Override

public void open(Map m, TopologyContext t, SpoutOutputCollector c) {

this.collector = collector;

try {

reader = new BufferedReader(new InputStreamReader(new FileInputStream("1.txt")));

} catch (FileNotFoundException e) {

e.printstacktrace();

}

}



/**

   * nextTuple()方法是任何Spout实现的核心。 Storm调用这个方法，向输出的    * collector发出tuple。

   */

@Override

public void nextTuple() {

try {

number = reader.readLine();

if (number != null) {

collector.emit(new Values(integer.parseInt(number)));

}

Utils.sleep(1000);

} catch (IOException e) {

e.printStackTrace();

}

}

/**

 * declareOutputFields是在Icomponent接口中定义的，所有Storm的组件（spout和bolt）都必须实现这个接口

 * 用于告诉Storm流组件将会发出那些数据流，每个流的tuple将包含的字段

 */

@Override

public void declareOutputFields(OutputFieldsDeclarer declarer) {

declarer.declare(new Fields("number"));

}



}

//BaseRichBolt是IComponent和IBolt接口的实现

    //继承这个类，就不用去实现本例不关心的方法



public static class SumBolt extends BaseRichBolt {



private static final long serialVersionUID = 1L;



private int sum = 0;

/**

     * prepare()方法类似于ISpout 的open()方法。

     * 这个方法在blot初始化时调用，可以用来准备bolt用到的资源,比如数据库连接。

     * 本例子和SentenceSpout类一样,SplitSentenceBolt类不需要太多额外的初始化,

     * 所以prepare()方法只保存OutputCollector对象的引用。

     */

@SuppressWarnings("rawtypes")

@Override

public void prepare(Map m, TopologyContext t, OutputCollector c) {



}

/**

     * SplitSentenceBolt核心功能是在类IBolt定义execute()方法，这个方法是IBolt接口中定义。

     * 每次Bolt从流接收一个订阅的tuple，都会调用这个方法。

     * 本例中,收到的元组中查找“sentence”的值,

     * 并将该值拆分成单个的词,然后按单词发出新的tuple。

     */

@Override

public void execute(Tuple input) {

Integer number = input.getIntegerByField("number");

sum += number;

System.out.println("--------------->" + sum);

}



@Override

public void declareOutputFields(OutputFieldsDeclarer d) {

// 这里是末端bolt，不需要发射数据流，这里无需定义

}



}



public static void main(String[] args) {

TopologyBuilder builder = new TopologyBuilder();



//设置spout组件需要的executor数量

Number parallelism_hint = 2;

//设置spout组件需要的task数量

Number number_task = 4;

builder.setSpout("sum-spout", new SumSpout(), parallelism_hint).setNumTasks(number_task);

//在以上代码中我们为SumSpout配置了2个初始执行线程（executor）和4个 //关联任务（task），这样每一个执行线程会运行2个任务，

//如果在设置spout的时候没有执行task的数量，那么每个exector的task //数量默认指定为1



builder.setBolt("sum-bolt", new SumBolt()).globalGrouping("sum-spout");

Config config = new Config();



//设置topology在集群中运行所需要的的工作进程的数量，默认一个topology //使用一个工作进程worker

config.setNumWorkers(2);



StormTopology topology = builder.createTopology();

if (args != null && args.length > 0) {

// 往storm集群提交作业

try {

StormSubmitter.submitTopology("sum-topology", config, topology);

} catch (AlreadyAliveException e) {

e.printStackTrace();

} catch (InvalidTopologyException e) {

e.printStackTrace();

} catch (authorizationException e) {

e.printStackTrace();

}

} else {

System.out.println("本地");



//建立本地集群，利用localcluster，storm在程序启动时会在本地自动建 //立一个集群，不需要用户自己再搭建，方便本地开发和调试

LocalCluster cluster = new LocalCluster();



cluster.submitTopology("sum-topology", config, topology);

}

}



}

相关阅读

分布式系统简介

分布式系统简介   工作中发现大数据技术的底层还是分布式系统，那么重新拾起，总结下~ 一、分布式系统简介 分布式系统是若干独立

NoSQL-Tidis支持分布式事务，兼容redis协议，使用tikv存储

项目repo地址 https://github.com/yongman/tidis Tidis是分布式数据库，支持redis协议，多种数据结构支持，编写语言为golang。 Tidis

淘宝技术发展（分布式时代：服务化）

在另外一个方面，还有两个新兴的业务，也对系统基础功能的拆分提出了要求。在那个时候，我们做了淘宝旅行（trip.taobao.com）和淘宝彩票（cai

分布式缓存击穿（布隆过滤器 Bloom Filter）

缓存中无值（未宕机） 互斥锁 缓存永不过期 缓存宕机 白名单 布隆过滤器 代码实现 前面的文章介绍了缓存的分类和使用的场景

饿了么分布式服务治理及优化经验

导读：高可用架构 7 月 30 日在上海举办了『互联网架构的基石』专题沙龙，进行了闭门私董会研讨及对外开放的四个专题的演讲，期望能促