概述

上一篇【 storm开发环境搭建】博文连接：https://my.oschina.net/u/2342969/blog/878765

本篇会深入理解Streams,欢迎同志(此同志非彼同志)们通过私信/评论等方式共同学习了解.

Streams是storm中一个核心的概念，它是在分布式并行处理和创建的无限序列元组，Streams通过给流元组中字段命名来定义，默认情况下，元组可以包含整型，长整型，短整型，字节，字符串，布尔型，双精度浮点型，单精度浮点型，字节数组，也可以自定义序列化类型。

下面共同学习一下 Tuple(元组)、OutputFieldsDeclarer、元组中动态类型以及序列器

Tuple(元组)

Tuple是storm中主要的数据结构，是storm中使用的基本单元、元组。元组是一个值列表，其中，每个值可以是任意类型。动态类型的元组不需要被定义，元组有类似 getInteger 和getString的帮助方法，无须手动转换结果类型。

storm需要知道如何序列化所有的值，默认情况下，storm知道如何序列化简单类型，比如字符串、字节数组，如果想使用复杂的对象，则需要注册实现一个该类型的序列器。

在storm中tuple接口集成了Iuple接口，实现类为TupleImpl。

OutputFieldsDeclarer

tuple的数据结构类似于map的键值对，其中键定义在OutputFieldsDeclarer方法的Fields对象中。

通过以下例子，可以帮助大家更好的理解：

//例2-2 src/main/java/bolts/WordNormalizer.java
package bolts;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
public class WordNormalizer implements IRichBolt {
    private OutputCollector collector;
    public void cleanup(){}
    /**
     * *bolt*从单词文件接收到文本行，并标准化它。
     * 文本行会全部转化成小写，并切分它，从中得到所有单词。
     */
    public void execute(Tuple input){
        String sentence = input.getString(0);
        String[] words = sentence.split(" ");
        for(String word : words){
            word = word.trim();
            if(!word.isEmpty()){
                word=word.toLowerCase();
                //发布这个单词
                List a = new ArrayList();
                a.add(input);
                collector.emit(a,new Values(word));
            }
        }
        //对元组做出应答
        collector.ack(input);
    }
    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
        this.collector=collector;
    }

    /**
     * 这个*bolt*只会发布“word”域
     */
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("word"));
    }

    @Override
    public Map<String, Object> getComponentConfiguration() {
        return null;
    }
}

创建了发送一个字段（“word”）的Bolt，此时tuple的键为“word”，值为execute方法中发送的Values对象。

序列器

本次介绍的storm0.6.0（含）及后续版本中如何使用序列器，storm在0.6.0之前使用不同的序列器，这里就不介绍老版本的。

tuple可以由任意类型组合而成，因为storm是分布式的，所以它需要知道在task间如何序列化和反序列化数据的。

storm使用Kryo进行序列化，Kryo是java开发中一个快速灵活序列器。默认情况下，storm可以序列化基础类型，比如字符串，字节，数组，ArrayList, HashMap, HashSet和 Clojure 集合类型，如果需要使用其他类型，需要自定义序列器。

动态类型

在元组中没有对应类型的字段。在字段中放入对象和storm动态序列化数据，得到序列化接口前，我们了解一下为什么storm元组是动态类型。

增加静态类型会大大增加storm API的复杂性, Hadoop 中,使用静态类型的键和值,在使用是需要大量的注释,对于hadoop API使用以及类型安全是一个不值得的负担,动态类型使用起来会很简单。

此外，storm 元组没有一个合理的方式使用静态类型，假如一个bolt订阅了多个流，那些流中元组会有不同类型传输在字段中。当一个bolt在execute方法接收元组，可以接收任何流的元组，就会有很多类型的元组。这样在一个bolt中，就需要为每个类型的tuple生命不同的方法订阅，显然，storm选择了简单方式，使用动态类型。

最后，另外使用动态类型的理由是storm可以直接被 Clojure 和 JRuby 这类动态类型的语言使用。

自定义序列器

综上所述，storm 使用Kryo 作为序列器。为了实现自定义序列器，就需要用Kryo注册一个新的序列器，

在Kryo的Github主页： https://github.com/EsotericSoftware/kryo，有更加详细的介绍，这里仅做一下简单介绍。

增加自定义序列器，需要在拓扑配置中添加“ topology.kryo.register ”属性，它可以配置一组序列器列表，每个序列器可以选择一下两种方式之一：

用类名注册，在例子中，storm会使用 Kryo 的“FieldsSerializer”序列化类--这可能不是最好的方式，在Kryo文档中有很多种方式。
从一个类名映射到实现了“ com.esotericsoftware.kryo.Serializer ”的注册器。

例子如下：

topology.kryo.register:
  - com.mycompany.CustomType1
  - com.mycompany.CustomType2: com.mycompany.serializer.CustomType2Serializer
  - com.mycompany.CustomType3

“com.mycompany.CustomType1“ 和“com.mycompany.CustomType3“ 使用“FieldsSerializer”序列化，反之，“com.mycompany.CustomType2“会使用”com.mycompany.serializer.CustomType2Serializer“ 序列化。

storm提供了在拓扑配置中注册序列器的助手，Config类调用registerSerialization方法可以将一个序列器放入配置中。其中有一个高级设置“Config.TOPOLOGY_SKIP_MISSING_KRYO_REGISTRATIONS”，如果把它设置为true，storm将会忽略在类路径无有效代码的序列器，否则，storm找不到序列器，将会排除异常。当在集群中运行了很多使用了不同序列器的拓扑，想通过“storm.yaml”文件为所有拓扑配置好序列器，它就非常有用的。