欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页  >  科技

初识storm分布式流计算

程序员文章站 2022-05-13 16:31:13
在2011年Storm开源之前,由于Hadoop的火红,整个业界都在喋喋不休地谈论大数据。Hadoop的高吞吐,海量数据处理的能力使得人们可以方便地地处理海量数据。但是,Hadoop的缺点...

在2011年Storm开源之前,由于Hadoop的火红,整个业界都在喋喋不休地谈论大数据。Hadoop的高吞吐,海量数据处理的能力使得人们可以方便地地处理海量数据。但是,Hadoop的缺点也和它的优点同样鲜明——延迟大,响应缓慢,运维复杂。

有需求也就有创造,在Hadoop基本奠定了大数据霸主地位的时候很多的开源项目都是以弥补Hadoop的实时性为目标而被创造出来。而在这个节骨眼上

Storm横空出世了。

Storm带着流式计算的标签华丽丽滴出场了,其实storm相对来说比hadoop复杂多了.如果你学会storm以后再去看hadoop那是非常简单的,看看它的一些卖点

分布式系统:可横向拓展,现在的项目不带个分布式特性都不好意思开源。运维简单:Storm的部署的确简单。虽然没有Mongodb的解压即用那么简单,但是它也就是多安装两个依赖库而已。高度容错:模块都是无状态的,随时宕机重启。无数据丢失:Storm创新性提出的ack消息追踪框架和复杂的事务性处理,能够满足很多级别的数据处理需求。不过,越高的数据处理需求,性能下降越严重。多语言:实际上,Storm的多语言更像是临时添加上去似的。因为,你的提交部分还是要使用Java实现. 下面我们来对比一下hadoop与storm

初识storm分布式流计算

但是他们没有谁强谁弱,应为侧重的方向不一样,hadoop侧重离线海量数据的分析,而storm需要实时计算分析,比如交通流量的调查,

淘宝或者京东购买商品时候的实时推荐等等 都可以通过storm实现.Storm主要分为两种组件Nimbus和Supervisor。

这两种组件都是快速失败的,没有状态。任务状态和心跳信息等都保存在Zookeeper上的,提交的代码资源都在本地机器的硬盘上。

Nimbus负责在集群里面发送代码,分配工作给机器,并且监控状态。全局只有一个。Supervisor会监听分配给它那台机器的工作,根据需要启动/关闭工作进程Worker。每一个要运行Storm的机器上都要部署一个,并且,按照机器的配置设定上面分配的槽位数。Zookeeper是Storm重点依赖的外部资源。Nimbus和Supervisor甚至实际运行的Worker都是把心跳保存在Zookeeper上的。Nimbus也是根据Zookeerper上的心跳和任务运行状况,进行调度和任务分配的。Storm提交运行的程序称为Topology。Topology处理的最小的消息单位是一个Tuple,也就是一个任意对象的数组。Topology由Spout和Bolt构成。Spout是发出Tuple的结点。Bolt可以随意订阅某个Spout或者Bolt发出的Tuple。Spout和Bolt都统称为component。

我们来看张经典的Topology的流程图就明白了

初识storm分布式流计算

在看张storm的架构图

初识storm分布式流计算

 

 

可能看到这或许你大概知道怎么工作的了,

下面看下我们的hell world

首先引入依赖

  
        org.apache.storm
        storm-core
        1.1.0
      

看下我们的Topology

 public static void main(String[] args) throws InterruptedException {
        //初始化top构造器
        TopologyBuilder top=new TopologyBuilder();
        //设置spout
        top.setSpout("spout",new TestSpout());
        //设置bolt
        top.setBolt("print-bolt",new TestBoot()).shuffleGrouping("spout");
        top.setBolt("write-bolt", new WriteBolt()).shuffleGrouping("print-bolt");
        //本地模式
        LocalCluster cluster=new LocalCluster();
        //提交top图
        cluster.submitTopology("tag",new Config(),top.createTopology());
        Thread.sleep(10000);
        cluster.killTopology("tap");
        cluster.shutdown();

    }
下面是Spout
public class TestSpout extends BaseRichSpout {
    private static final long serialVersionUID = 1L;
    private SpoutOutputCollector collector;

    private static final Map map = new HashMap();

    static {
        map.put(0, "java");
        map.put(1, "php");
        map.put(2, "groovy");
        map.put(3, "python");
        map.put(4, "ruby");
    }

    @Override
    public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
        //对spout进行初始化
        this.collector = collector;
        System.out.println("++++++++++++++++++++++++++++++");
        //System.out.println(this.collector);
    }


    /**
     * 方法名称:轮询tuple

     * 概要说明:

     * @see
     */
    @Override
    public void nextTuple() {
        //随机发送一个单词
        final Random r = new Random();
        int num = r.nextInt(5);
        System.out.println("----------------------------");
        try {
            Thread.sleep(500);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        this.collector.emit(new Values(map.get(num)));
    }

    /**
     * 方法名称:declarer声明发送数据的field

     * 概要说明:

     * @see
     */
    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        System.out.println("=============================");
        //进行声明
        declarer.declare(new Fields("print"));
    }

}
TestBoot
public class TestBoot extends BaseRichBolt {

    OutputCollector _collector;


    @Override
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
        _collector=outputCollector;
    }

    @Override
    public void execute(Tuple tuple) {
        String spout = tuple.getStringByField("print");
        System.out.println("1111111111111"+spout);
        //发射字段
        _collector.emit(new Values(spout));
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        //这是下游接收者
        outputFieldsDeclarer.declare(new Fields("write"));
    }
}

writeBolt
public class WriteBolt extends BaseRichBolt {

	private static final long serialVersionUID = 1L;

	private static final Log log = LogFactory.getLog(WriteBolt.class);
	
	private FileWriter writer ;
	@Override
	public void execute(Tuple input) {
		//获取上一个组件所声明的Field
		String text = input.getStringByField("write");
		System.out.println(text);
		try {
			if(writer == null){
				if(System.getProperty("os.name").equals("Windows 10")){
					writer = new FileWriter("D:\\099_test\\" + this);
				} else if(System.getProperty("os.name").equals("Windows 8.1")){
					writer = new FileWriter("D:\\099_test\\" + this);
				} else if(System.getProperty("os.name").equals("Windows 7")){
					writer = new FileWriter("D:\\099_test\\" + this);
				} else if(System.getProperty("os.name").equals("Linux")){
					System.out.println("----:" + System.getProperty("os.name"));
					writer = new FileWriter("/usr/local/temp/" + this);
				}
			}
			log.info("【write】: 写入文件");
			writer.write(text);
			writer.write("\n");
			writer.flush();
			
		} catch (Exception e) {
			e.printStackTrace();
		}
	}


	@Override
	public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

	}

	@Override
	public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {

	}

}

我们的Topology图的流程是这样的由TestSpout 流向 TestBoot 然后在流向 WriteBolt ,TestBoot负责在控制台打印,WriteBolt 负责往磁盘写入 ,下面使我们的运行结果

初识storm分布式流计算

通过Spout随机给TestBoot发送单词,然后经过WriteBolt进行写入到磁盘.并且启动的是本地模式.如果想在服务器上的storm环境中运行Topology 需要改成集群模式

 try {
            //开启集群模式
            StormSubmitter.submitTopology("top1", new Config(), top.createTopology());
        } catch (AlreadyAliveException e) {
            e.printStackTrace();
        } catch (InvalidTopologyException e) {
            e.printStackTrace();
        } catch (AuthorizationException e) {
            e.printStackTrace();
        }

本文章重点是要让大家对Storm有个初步的认识,希望能帮到大家