概述

上一篇【 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对象中。

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

1//例2-2 src/main/java/bolts/WordNormalizer.java
2package bolts;
3import org.apache.storm.task.OutputCollector;
4import org.apache.storm.task.TopologyContext;
5import org.apache.storm.topology.IRichBolt;
6import org.apache.storm.topology.OutputFieldsDeclarer;
7import org.apache.storm.tuple.Fields;
8import org.apache.storm.tuple.Tuple;
9import org.apache.storm.tuple.Values;
10
11import java.util.ArrayList;
12import java.util.List;
13import java.util.Map;
14public class WordNormalizer implements IRichBolt {
15    private OutputCollector collector;
16    public void cleanup(){}
17    /**
18     * *bolt*从单词文件接收到文本行，并标准化它。
19     * 文本行会全部转化成小写，并切分它，从中得到所有单词。
20     */
21    public void execute(Tuple input){
22        String sentence = input.getString(0);
23        String[] words = sentence.split(" ");
24        for(String word : words){
25            word = word.trim();
26            if(!word.isEmpty()){
27                word=word.toLowerCase();
28                //发布这个单词
29                List a = new ArrayList();
30                a.add(input);
31                collector.emit(a,new Values(word));
32            }
33        }
34        //对元组做出应答
35        collector.ack(input);
36    }
37    public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
38        this.collector=collector;
39    }
40
41    /**
42     * 这个*bolt*只会发布“word”域
43     */
44    public void declareOutputFields(OutputFieldsDeclarer declarer) {
45        declarer.declare(new Fields("word"));
46    }
47
48    @Override
49    public Map<String, Object> getComponentConfiguration() {
50        return null;
51    }
52}

创建了发送一个字段（“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 ”的注册器。

例子如下：

1topology.kryo.register:
2  - com.mycompany.CustomType1
3  - com.mycompany.CustomType2: com.mycompany.serializer.CustomType2Serializer
4  - 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”文件为所有拓扑配置好序列器，它就非常有用的。

JAVA序列器

当storm遇到一个没有序列器注册的类型,它可能会使用java序列器,如果此类型也无法被java序列器序列化,storm就会报出一个错误。

需要注意的是，无论在CPU消耗方面还是序列化对象的大小， java序列器都是非常耗费资源的。故在生产上使用拓扑的话，强烈建议使用自定义序列器。java序列器在那里，是为了容易设计新的原型。

通过设置“Config.TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION”项为false，可以关闭java序列器。

特定组件序列化注册

Storm 0.7.0 可以设置特殊组件配置，当然，如果一个组件定义一个序列化，这个序列器需要对其他bolt可用，否则其他bolt将无法接收那个组件的消息。

一个拓扑被提交，一组序列器会被拓扑选择为所有组件发送消息使用，这是通过特殊组件序列化注册信息与普通组件序列化注册信息合并实现。当为同一个类注册了多个序列化时，序列器会任意选择一个。

如果两个特定组件序列器有冲突，则会强制一个特定的类做序列器，只需在拓扑特定配置定义想要的序列器，拓扑特定配置优先于序列器注册的特定组件。

storm从入门到放弃教程（5）

概述