19. Storm Topology开发

作者: 奉先 | 来源:发表于2017-09-27 17:13 被阅读260次

一. wordCount Topology开发：

1.spout数据收集器（SentenceSpout类）：

有两种方法来开发spout类，第一种是实现backtype.storm.topology.IRichSpout接口，第二种是继承backtype.storm.topology.base.BaseRichSpout类。
其中，IRichSpout接口提供了更多的一些需要实现的方法，BaseRichSpout类只提供了3个需要实现的方法。

    @Override
    public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
        // TODO Auto-generated method stub
        this.collector = collector;
    }
    @Override
    public void close() {
        // TODO Auto-generated method stub
        
    }
    @Override
    public void activate() {
        // TODO Auto-generated method stub
        
    }
    @Override
    public void deactivate() {
        // TODO Auto-generated method stub
        
    }
    @Override
    public void nextTuple() {
        // TODO Auto-generated method stub
        
    }
    @Override
    public void ack(Object msgId) {
        // TODO Auto-generated method stub
        
    }
    @Override
    public void fail(Object msgId) {
        // TODO Auto-generated method stub
        
    }
    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        // TODO Auto-generated method stub
        
    }
    @Override
    public Map<String, Object> getComponentConfiguration() {
        // TODO Auto-generated method stub
        return null;
    }

上边的这些方法中，有几个比较重要，需要实现。
1.nextTuple()：
在该方法中，编写从数据源获取数据的逻辑。该方法程序循环调用，collector.emit()向后边的bolt发射数据。
2.declareOutputFields()：
该方法声明向后边发射的记录的字段名称。
3.tuple
collector.emit()方法发射的内容是Tuple，类型为List<Object> tuple。 tuple元组是一系列key，value对的集合。例如：(a:a_value,b:b_value,c:c_value,...,n:n_value)。其中，collector.emit(new Values())声明的是tuple的value值，而declarer.declare(new Fields())声明的是tuple的key值，两者是一一对应的（假如new Values(val1,val2)，那么，declarer.declare(new Fields(key1,key2))也需要声明2个值，并且key1对一个val1，key2对应val2）。
4.open()：
该方法是初始化方法，将会第一个被调用，一般，我们可以在该方法内实例化定义的类。

2.bolt组件（SplitBolt、CountBolt类）：

开发bolt组件，需要实现backtype.storm.topology.IRichBolt接口，或者继承类backtype.storm.topology.base.BaseRichBolt。
下面几个方法比较重要：
1.prepare()
初始化方法，将会第一个被调用，一般，我们可以在该方法内实例化定义的类。
2.execute()
循环调用，被动执行,前面数据来源向该bolt发射tuple的时候，就会调用execute方法。
3.declareOutputFields
与spout相同。

3.Topology驱动类（WordsToplogy类）：

向集群提交Topology，需要使用类backtype.storm.topology.TopologyBuilder。TopologyBuilder类可以配置spout、bolt组件的记录发射关系（前后依赖关系，例如：spout --> bolt1 -->bolt2等）。

TopologyBuilder builder = new TopologyBuilder();        
builder.setSpout("Spout", new SentenceSpout());
builder.setBolt("SplitBolt", new SplitBolt()).shuffleGrouping("Spout");
builder.setBolt("CountBolt", new CountBolt())
 .fieldsGrouping("SplitBolt", new Fields("Word"));

所谓的grouping策略就是在Spout与Bolt、Bolt与Bolt之间传递Tuple的方式。Grouping分组策略主要有以下几种：
1.shuffleGrouping：随机分组。将流分组定义为混排。这种混排分组意味着来自Spout的输入将混排，或随机分发给此Bolt中的任务。shuffle grouping对各个task的tuple分配的比较均匀。
2.fieldsGrouping：按照字段分组。Storm能保证所有相同Field值的数据到达的是相同的Blot，但是不保证一个Blot只处理一个值域。这对于分组统计的应用来说是比较重要的，如果分组不正确的话会造成统计出错。
3.globalgrouping：全局分组，前面组件的数据,全部只会往该组件的其中一个上传送。
4.allGrouping:广播发送，即每一个Tuple，每一个Bolt都会收到。

4.Storm集群参数配置：

对于storm集群的参数，可以通过Config对象来配置。

Config conf = new Config();
conf.setMaxSpoutPending(10);

也可以通过conf.put(key, value)来配置xml文件中的参数。

5.单机运行或者提交storm集群：

单机提交topology（主要用于提交集群前的测试，非常重要）。使用LocalCluster类来提交单机测试topology：

LocalCluster local = new LocalCluster();
local.submitTopology("LocalTest", conf, builder.createTopology());

集群提交topology：

StormSubmitter.submitTopology("WordCount", conf, builder.createTopology());

6.代码示例：

下面的代码实现了一个词频统计的storm实例，功能非常简单，随机发送sentence并拆分统计单词。
https://github.com/neil-ma/storm-pmpa/tree/master/storm-pirate/src/main/java/com/pmpa/storm/words

二. Topology并发控制：

1.并发控制组件：

Storm的并发度最终表示的Task的并发度。Storm执行架构有三个层次 Worker -> Executor -> Task。配置以上3个组件的数量来控制并发度。
Worker进程：针对具体的Topology，worker上只运行与之相关的Topology，一个worker进程上可以启动多个executor线程。
Executor线程：针对具体的task（spout、bolt），一个Executor线程上可以跑多个task，默认一个Executor运行一个task。
Task：指定多个task来运行spout或者bolt组件。

2.参数配置：

1.Worker进程数量：
通过Config设置： conf.setNumWorkers(4); // 设置worker个数为4
Supervisor进程负责启动worker，假如有3个supervior，这3个supervisor会平均配置4个Worker，例如： 2 1 1 。
2.Executor个数：
在构造Topology时，在setSpout()或者setBolt()方法中设定executor的数量。例如下例子：

builder.setBolt("SplitBolt", new SplitBolt(),3).shuffleGrouping("Spout")

代码表示，需要启动3个executor来运行SplitBolt。需要注意的是，这里表示一共有3个executor，而不是每个worker上运行3个executor。假如说，config的配置一共有2个worker，那么分配的结果就是一个worker上执行2个executor，另一个worker上执行1个executor。后边的task配置也是一样的道理。
3.task个数
task的数量由setNumTasks()方法确定，例如下边的定义：

builder.setBolt("SplitBolt", new SplitBolt(),3).shuffleGrouping("Spout")
  .setNumTasks(6);

上边代码表示3个executor共执行6个task，storm会平均分配一个executor执行2个task（系统自动做到尽量均匀）。如果不指定setNumTasks()方法，默认1个Executor运行一个Task，上边代码如果不指定setNumTasks()方法会有3个Task执行。

三. Storm消息可靠性保障机制：

对于某些实时大数据应用，例如银行的实时数据、交管部门的实时数据等，需要保证数据的可靠性，在实施这类应用时，就需要开启storm的消息可靠性保障机制。消息可靠性保障机制实际上就是Storm需要对spout发送的每一条消息是否被后续的bolt成功处理完成有一条反馈。

1.原理和机制：

1.ack机制：
为了保证storm的每条记录都能正确处理，Storm会对Spout发送的每一个tuple进行跟踪。这里面包括ack/fail的处理，一个tuple处理成功是指这个Tuple以及这个Tuple产生的所有Tuple都被成功处理, 会调用spout的ack方法；失败是指这个Tuple或这个Tuple产生的所有Tuple中的某一个tuple处理失败, 则会调用spout的fail方法；在处理tuple的每一个bolt都会通过OutputCollector来告知storm, 当前bolt处理是否成功。
2.ack原理：
Storm中有个特殊的task名叫acker，他们负责跟踪spout发出的每一个Tuple的Tuple树。当acker（框架自启动的task）发现一个Tuple树已经处理完成了，它会发送一个消息给产生这个Tuple的那个task。

2.实现：

1.spout处理：
（1）spout往后发射tuple时，需要指定一个msgId。

2.bolt处理：
（1）bolt处理接收到tuple，如果还需要继续往后边的bolt发射，需要追溯前边的tuple（这么做的目的是构建Tuple树）

collector.emit(input,new Values(word));

（2）处理完bolt，一定要调用collector的ack方法，

四. Trident介绍和实现：

1. 问题：

前边介绍的基础的storm都是逐条处理数据的（一个tuple、一个tuple处理）。在生产环境中，一般都是Kafka + Storm + HBase/Redis 架构处理实时数据。如果只是逐条处理的话，对下游数据库（HBase、Redis）的压力就会非常大。
Trident是Storm提供的解决方案，一个批次一个批次处理实时数据，其中一个批次封装了多条tuple。Trident能够提高数据处理效率和性能，同时也减小了对后端数据库的压力。因为Trident是以批次为单位来处理数据的，所以这里就涉及到事务的问题。Trident中已经封装了事务管理、状态管理的功能（框架帮我们自动实现），而且还封装了一系列的常用操作，链式调用。真正实现流式处理数据。
Storm从0.7版本开始引入事务管理，之前版本中提供的Transactional Topology API已经废弃不用了。

2. Storm事务管理：

Storm事务管理分为3个层次：
（1）No Transactional：
不进行事务管理。一个批次中的tuple可能有的成功，有的失败，不限制一致性。tuple处理成功次数可能不止一次，同一个tuple可能在多个批次中处理，并且都成功，也可能一次都不成功。
（2）Transactional ：
保证tuple只会在一个批次中出现，即使失败重试，tuple的批次号还是不变的，同一个tuple保证最多成功一次。
（3）Opaque Transactional：
不透明事务，和第2种类似。相比于第2种，提供了容错的机制。某些tuple在某个批次中处理失败后，可以在另外一个批次里处理成功（失败后，将该tuple转到另外一个批次中处理），但不会成功多次。

3. Storm事务原理：

（1）将多条tuple封装成一个批次，并且给该批次指定一个唯一的批次号（batchId）。
（2）后边组件处理数据按照批次先后顺序处理（前边的批次更新后，才能处理后边的批次），结果的更新，一定是前面的批次更新成功后才能进行后面的批次结果更新。

4. WordCountTridentTopology实现：

TridentTopology需要开发自己的spout（以前是逐条发送tuple，现在的需要将多条tuple封装成一个batch发送），自己的function（在trident中不叫bolt，而是function，实现的功能与bolt一样），下面实现了一个最简单的实例：
https://github.com/neil-ma/storm-pmpa/tree/master/storm-pirate/src/main/java/com/pmpa/storm/wordstrident

五. Trident编程：

1. 编写Trident Spout:

编写Trident Spout需要自行实现将tuple打包成batch的逻辑。