Apache Storm入门学习

=Start=

缘由：

学习、提高需要

正文：

参考解答：

一、什么是Apache Storm？

Apache Storm是一个分布式实时大数据处理系统。Storm设计用于在容错和水平可扩展方法中处理大量数据。它是一个流数据框架，具有很高的吞吐率。虽然Storm是无状态的，它通过Apache ZooKeeper管理分布式环境和集群状态。它很简单，您可以并行地对实时数据执行各种操作。Apache Storm易于设置和操作，并且它保证每个消息将通过拓扑至少处理一次。

二、Apache Storm vs. Hadoop

Hadoop和Storm框架都用于分析大数据。两者互补，在某些方面有所不同。Apache Storm执行除持久性之外的所有操作，而Hadoop在所有方面都很好，但滞后于实时计算。下表比较了Storm和Hadoop的属性。

Storm	Hadoop
实时流处理	批量处理
无状态	有状态
主/从架构与基于ZooKeeper的协调。主节点称为nimbus，从属节点是主管。	具有/不具有基于ZooKeeper的协调的主 – 从结构。主节点是作业跟踪器，从节点是任务跟踪器。
Storm流过程在集群上每秒可以访问数万条消息。	Hadoop分布式文件系统（HDFS）使用MapReduce框架来处理大量的数据，需要几分钟或几小时。
Storm拓扑运行直到用户关闭或意外的不可恢复故障。	MapReduce作业按顺序执行并最终完成。
两者都是分布式和容错的
如果nimbus / supervisor死机，重新启动使它从它停止的地方继续，因此没有什么受到影响。	如果JobTracker死机，所有正在运行的作业都会丢失。

Storm：分布式实时计算，强调实时性，常用于实时性要求较高的地方；
Hadoop：分布式批处理计算，强调批处理，常用于对已经在的大量数据进行挖掘、分析；

三、Apache Storm的核心概念

下图描述了Apache Storm的核心概念。

核心理念

Storm 集群的输入流由一个被称作 spout 的组件管理，spout 把数据传递给 bolt， bolt 要么把数据保存到某种存储器，要么把数据传递给其它的 bolt。你可以想象一下，一个 Storm 集群就是在一连串的 bolt 之间转换 spout 传过来的数据：

data-source -> spout -> bolt[ -> bolt ...] -> data-storage

现在让我们仔细看看Apache Storm的组件：

组件	描述
Tuple	Tuple是Storm中的主要数据结构。它是有序元素的列表。默认情况下，Tuple支持所有数据类型。通常，它被建模为一组逗号分隔的值，并传递到Storm集群。
Stream	流是元组的无序序列。
Spouts	流的源。通常，Storm从原始数据源（如Twitter Streaming API，Apache Kafka队列，Kestrel队列等）接受输入数据。否则，您可以编写spouts以从数据源读取数据。“ISpout”是实现spouts的核心接口，一些特定的接口是IRichSpout，BaseRichSpout，KafkaSpout等。
Bolts	Bolts是逻辑处理单元。Spouts将数据传递到Bolts和Bolts过程，并产生新的输出流。Bolts可以执行过滤，聚合，加入，与数据源和数据库交互的操作。Bolts接收数据并发射到一个或多个Bolts。 “IBolt”是实现Bolts的核心接口。一些常见的接口是IRichBolt，IBasicBolt等。

拓扑

Spouts和Bolts连接在一起，形成拓扑结构。实时应用程序逻辑在Storm拓扑中指定。简单地说，拓扑是有向图，其中顶点是计算，边缘是数据流。

简单拓扑从spouts开始。Spouts将数据发射到一个或多个Bolts。Bolt表示拓扑中具有最小处理逻辑的节点，并且Bolts的输出可以发射到另一个Bolts作为输入。

Storm保持拓扑始终运行，直到您终止拓扑。Apache Storm的主要工作是运行拓扑，并在给定时间运行任意数量的拓扑。

任务

现在你有一个关于Spouts和Bolts的基本想法。它们是拓扑的最小逻辑单元，并且使用单个Spout和Bolt阵列构建拓扑。应以特定顺序正确执行它们，以使拓扑成功运行。Storm执行的每个Spout和Bolt称为“任务”。简单来说，任务是Spouts或Bolts的执行。在给定时间，每个Spout和Bolt可以具有在多个单独的螺纹中运行的多个实例。

进程

拓扑在多个工作节点上以分布式方式运行。Storm将所有工作节点上的任务均匀分布。工作节点的角色是监听作业，并在新作业到达时启动或停止进程。

四、Apache Storm的集群架构

Apache Storm的主要亮点是，它是一个容错，快速，没有“单点故障”（SPOF）分布式应用程序。我们可以根据需要在多个系统中安装Apache Storm，以增加应用程序的容量。

让我们看看Apache Storm集群如何设计和其内部架构。下图描述了集群设计：

动物园管理员框架

Apache Storm有两种类型的节点，Nimbus（主节点）和Supervisor（工作节点）。Nimbus是Apache Storm的核心组件。Nimbus的主要工作是运行Storm拓扑。Nimbus分析拓扑并收集要执行的任务。然后，它将任务分配给可用的supervisor。

Supervisor将有一个或多个工作进程。Supervisor将任务委派给工作进程。工作进程将根据需要产生尽可能多的执行器并运行任务。Apache Storm使用内部分布式消息传递系统来进行Nimbus和管理程序之间的通信。

组件	描述
Nimbus（主节点）	Nimbus是Storm集群的主节点。集群中的所有其他节点称为工作节点。主节点负责在所有工作节点之间分发数据，向工作节点分配任务和监视故障。
Supervisor（工作节点）	遵循指令的节点被称为Supervisors。Supervisor有多个工作进程，它管理工作进程以完成由nimbus分配的任务。
Worker process（工作进程）	工作进程将执行与特定拓扑相关的任务。工作进程不会自己运行任务，而是创建执行器并要求他们执行特定的任务。工作进程将有多个执行器。
Executor（执行者）	执行器只是工作进程产生的单个线程。执行器运行一个或多个任务，但仅用于特定的spout或bolt。
Task（任务）	任务执行实际的数据处理。所以，它是一个spout或bolt。
ZooKeeper framework（ZooKeeper框架）	Apache的ZooKeeper的是使用群集（节点组）自己和维护具有强大的同步技术共享数据之间进行协调的服务。Nimbus是无状态的，所以它依赖于ZooKeeper来监视工作节点的状态。 ZooKeeper的帮助supervisor与nimbus交互。它负责维持nimbus，supervisor的状态。

Storm是无状态的。即使无状态性质有它自己的缺点，它实际上帮助Storm以最好的可能和最快的方式处理实时数据。

Storm虽然不是完全无状态的。它将其状态存储在Apache ZooKeeper中。由于状态在Apache ZooKeeper中可用，故障的网络可以重新启动，并从它离开的地方工作。通常，像monit这样的服务监视工具将监视Nimbus，并在出现任何故障时重新启动它。

五、Apache Storm的入门实例

HelloWorldSpout.java

public class HelloWorldSpout extends BaseRichSpout{
    private SpoutOutputCollector collector;
    private int referenceRandom;
    private static final int MAX_RANDOM = 10;

    public HelloWorldSpout() {
        final Random rand = new Random();
        referenceRandom = rand.nextInt(MAX_RANDOM);
    }

    /*
     * declareOutputFields() => you need to tell the Storm cluster which fields this Spout emits within the
     *  declareOutputFields method.
     */
    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("sentence"));
    }

    /*
     * open() => The first method called in any spout is 'open'
     *           TopologyContext => contains all our topology data
     *           SpoutOutputCollector => enables us to emit the data that will be processed by the bolts
     *           conf => created in the topology definition
     */
    @Override
    public void open(Map conf, TopologyContext topologyContext, SpoutOutputCollector collector) {
        this.collector = collector;
    }

    /*
     * nextTuple() => Storm cluster will repeatedly call the nextTuple method which will do all the work of the spout.
     *  nextTuple() must release the control of the thread when there is no work to do so that the other methods have
     *  a chance to be called.
     */
    @Override
    public void nextTuple() {
        final Random rand = new Random();
        int instanceRandom = rand.nextInt(MAX_RANDOM);
        if(instanceRandom == referenceRandom){
            collector.emit(new Values("Hello World"));
        } else {
            collector.emit(new Values("Other Random Word"));
        }
    }
}

HelloWorldBolt.java

public class HelloWorldBolt extends BaseRichBolt {
    private int myCount = 0;

    /*
     * prepare() => on create
     */
    @Override
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
    }

    /*
     * execute() => most important method in the bolt is execute(Tuple input), which is called once per tuple received
     *  the bolt may emit several tuples for each tuple received
     */
    @Override
    public void execute(Tuple tuple) {
        String test = tuple.getStringByField("sentence");
        if(test == "Hello World"){
            myCount++;
            System.out.println("Found a Hello World! My Count is now: " + Integer.toString(myCount));
        }
    }

    /*
     * declareOutputFields => This bolt emits nothing hence no body for declareOutputFields()
     */
    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
    }
}

HelloWorldTopology.java

/**
 * Author: ashrith
 * Desc: setup the topology and submit it to either a local of remote Storm cluster depending on the arguments
 *       passed to the main method.
 */
public class HelloWorldTopology {
    /*
     * main class in which to define the topology and a LocalCluster object (enables you to test and debug the
     * topology locally). In conjunction with the Config object, LocalCluster allows you to try out different
     * cluster configurations.
     *
     * Create a topology using 'TopologyBuilder' (which will tell storm how the nodes area arranged and how they
     * exchange data)
     * The spout and the bolts are connected using 'ShuffleGroupings'
     *
     * Create a 'Config' object containing the topology configuration, which is merged with the cluster configuration
     * at runtime and sent to all nodes with the prepare method
     *
     * Create and run the topology using 'createTopology' and 'submitTopology'
     */
    public static void main(String[] args) throws AlreadyAliveException, InvalidTopologyException {
        TopologyBuilder builder = new TopologyBuilder();
        builder.setSpout("randomHelloWorld", new HelloWorldSpout(), 10);
        builder.setBolt("HelloWorldBolt", new HelloWorldBolt(), 1).shuffleGrouping("randomHelloWorld");
        Config conf = new Config();
        conf.setDebug(true);
        if(args!=null && args.length > 0) {
            conf.setNumWorkers(3);
            StormSubmitter.submitTopology(args[0], conf, builder.createTopology());
        } else {
            LocalCluster cluster = new LocalCluster();
            cluster.submitTopology("test", conf, builder.createTopology());
            Utils.sleep(10000);
            cluster.killTopology("test");
            cluster.shutdown();
        }
    }
}

参考链接：

Apache Storm核心概念
https://www.w3cschool.cn/apache_storm/apache_storm_core_concepts.html
Apache Storm集群架构
https://www.w3cschool.cn/apache_storm/apache_storm_cluster_architecture.html
Apache Storm工作实例
https://www.w3cschool.cn/apache_storm/apache_storm_working_example.html
Storm教程
http://www.tianshouzhi.com/api/tutorials/storm
sample hello world topology with pom
https://github.com/ashrithr/storm-helloworld
Storm 入门教程
http://www.dahouduan.com/2015/11/10/storm-tutorial/

=END=

2 9 月, 2017

Docker

KnowledgeBase

bolt, Hadoop, spout, Storm, topo