Storm的数据处理编程单元:Bolt 学习整理

　　Bolt是Topology中的数据处理的单元，也是Storm针对处理过程的编程单元。Topology中全部的处理都是在这些Bolt中完成的，编程人员能够实现自定义的处理过程，例如，过滤、函数、汇集、链接等计算。若是是复杂的计算过程，每每须要多个步骤和使用多个Bolt。

　　Bolt能够将数据项发送至多个数据流(Stream)。编程人员首先可使用OutputFieldsDeclarer类的declareStream()方法来声明多个流，指定数据将要发送到的流，而后使用SpoutOutputCollector的emit方法将数据发送。

　　当声明了一个Bolt的输入流后，能够从其余的组件中接收这些指定的流。当接收某个组件的全部流时，须要在程序中逐个声明接收的过程。InputDeclarer对象默认接收来自某组件默认的流。

//从名称为"1"的组件中接收默认的流。
declarer.shuffleGrouping("1")

 

IBolt 和 IComponent接口

IBolt接口:

//在组件的任务初被初始化时，由集群中的工做进程(worker)调用,prepare()用于实例化Bolt的已给运行时任务，被集群中的某一个进程调用，提供Bolt运行的环境。
//sormConf对象维护Storm中针对该Bolt的配置信息。(来自Topology);context对象是一个上下文对象，用于获取该组件运行时任务的信息。(例如Topology中该Bolt全部任务的位置，包括任务的id、组件id和输入输出信息等)
//collector对象用于从该Bolt发送数据项。数据项能够在任意时刻发送，包括调用open()和close()方法。

void prepare(java.util.Map stormConf,TopologyContext context,OutputCollector collector)

//接收一个数据项并处理
//该方法用来接收一个数据项(Tuple)，并能够将处理的结果做为新的数据项发送(emit),是Bolt须要实现的最重要的方法。
//参数imput是一个数据项对象，包含了众多的元数据(metadata)，包括它来自的组件、流、任务等。数据项中的值，能够经过Tuple类的getValue()方法得到。

void execute(Tuple input)

//在IBolt将关闭时调用  
void cleanup()

Tuple类的方法，这个类的对象做为execute()方法的输入。（方法举例: int size() ; int fieldIndex(java.lang.String field) ; ......)

方法众多，能够整理分为如下五类:

一、获取属性的方法。 (size()、fieldIndex()和contains()三个方法）

二、获取元数据的方法。(getMessageId()、getSourceComponent()、getSourceTask()、getSourceStreamId()和getSourceGlobalStreamid()方法)

其中MessageId是在数据项被建立时，经过必定的规则赋值的。

三、根据域获取值的方法。(getValue()和多个get具体数据类型的方法)

四、根据域的名称获取值的方法。(这一类包括getFields()、getValues()和select()方法)

五、获取Tuple的值或域列表的方法。(getFields()、getValues()和select()方法)

分别获取该数据项的全部域列表、值列表和值列表子集。

 

简单的案例:

class SplitSentence implements IRichBolt {
    private OutputCollector collector;

    public void prepare(Map conf,TopologyContext context,OutputCollector collector){
    this.collector = collector;
    }
    
    public void execute(Tuple tuple){
    String sentence = tuple.getString(0);
    for(String word : sentence.split(" "){
        collector.emit(new Values(word));
    }
}

public void cleanup(){
}

public void declareOutputFields(OutpuFieldsDeclarer declarer){
    declarer.declare(new Fields("word"));
    }
}

这里说下declareOutputFields()函数参数，声明了输出流的数据项的结构，也即Tuple的域。

结合上节给的Spout的示例，能够在Topology类的main函数中加入相关代码,增长Bolt。

Topology builder builder = new TopologyBuilder();
Builder.SetSpout ("SentenceGenSpout ",new TestWord Spout(),1);
builder.setBolt("splitBoult",new SplitSentence(),2).shuffleGrouping("sentenceGenSpout");