Storm 折腾杂记

作者: 一只很努力爬树的猫 | 来源:发表于2017-11-23 11:13 被阅读458次

Date: Nov 17-24, 2017

1. 目的

积累Storm为主的流式大数据处理平台对实时数据处理的相关技术
积累快捷的Storm部署、开发方式，例如Python和Java。

2. 阅读资料

3. 阅读笔记

3.1 Apache Storm官网

3.1.1 Storm主要结构概览

Storm主要结构

如上图所示，Storm是一个流数据处理平台。它与Hadoop相近，采用Map-Reduce的计算框架，区别在于Hadoop的worker在完成工作后被释放，而Storm的worker在完成工作后进入等待状态——等待“上级”分配下一个任务。

Storm的本质是定义一个计算的过程，类似于设计中的数据流图，即先定义数据处理的流程，再分模块实现数据处理的细节，结果由末端的节点返回或输出。

Storm的核心是Clojure编写、提供Java开发接口，核心离工业解主流编程语言(Java、C/C++)相对遥远，阿里巴巴的工程师团队用Java重写了Storm的核心，即为JStorm。

3.2 阿里巴巴JStorm

3.2.1 JStorm定位

JStorm 是一个分布式实时计算引擎。

JStorm 是一个类似Hadoop MapReduce的系统，用户按照指定的接口实现一个任务，然后将这个任务递交给JStorm系统，JStorm将这个任务跑起来，并且按7 * 24小时运行起来，一旦中间一个Worker 发生意外故障，调度器立即分配一个新的Worker替换这个失效的Worker。

因此，从应用的角度，JStorm应用是一种遵守某种编程规范的分布式应用。从系统角度， JStorm是一套类似MapReduce的调度系统。从数据的角度，JStorm是一套基于流水线的消息处理机制。

	JStorm	Hadoop
角色	Nimubs	JobTracker
	Supervisor	TaskTracker
	Worker	Child
应用名称	Topology	Job
编程接口	Spout/Bolt	Mapper/Reducer
“设计模式”	资本主义	恐怖主义

3.2.2 优点

在Storm和JStorm出现以前，市面上出现很多实时计算引擎，但自Storm和JStorm出现后，基本上可以说一统江湖：究其优点:

开发非常迅速：接口简单，容易上手，只要遵守Topology、Spout和Bolt的编程规范即可开发出一个扩展性极好的应用，底层RPC、Worker之间冗余，数据分流之类的动作完全不用考虑
扩展性极好：当一级处理单元速度，直接配置一下并发数，即可线性扩展性能
健壮强：当Worker失效或机器出现故障时， 自动分配新的Worker替换失效Worker
数据准确性：可以采用Ack机制，保证数据不丢失。如果对精度有更多一步要求，采用事务机制，保证数据准确。
实时性高： JStorm 的设计偏向单行记录，因此，在时延较同类产品更低

3.2.3 应用场景

JStorm处理数据的方式是基于消息的流水线处理，因此特别适合无状态计算，也就是计算单元的依赖的数据全部在接受的消息中可以找到，并且最好一个数据流不依赖另外一个数据流。

因此，常常用于:

日志分析，从日志中分析出特定的数据，并将分析的结果存入外部存储器如数据库。目前，主流日志分析技术就使用JStorm或Storm
管道系统，将一个数据从一个系统传输到另外一个系统，比如将数据库同步到Hadoop
消息转化器，将接受到的消息按照某种格式进行转化，存储到另外一个系统如消息中间件
统计分析器，从日志或消息中，提炼出某个字段，然后做count或sum计算，最后将统计值存入外部存储器。中间处理过程可能更复杂。
实时推荐系统，将推荐算法运行在jstorm中，达到秒级的推荐效果

3.2.4 基本概念

[站外图片上传中...(image-96ea41-1511406796108)]

Spout (中文意为水龙头)即数据的来源、出水口，来源可以是Kafka、DB、HBase、HDFS等。
Bolt(中文意为插销)即数据流向过程中的关键点、数据流处理点。
Topology(中文意为拓扑结构)即上述图中所示的数据处理流程形成的数据流网络结构。

3.2.5 组件接口

Spout组件接口：nextTuple 拉取下一条消息，执行时JStorm框架回不停调用该接口从数据源拉取数据发往Bolt。
Bolt组件接口：execute 执行处理逻辑

3.2.6 调度和执行

对于一个Topology，JStorm调度一个/多个Worker (每个Worker对应操作系统的进程)，分布到集群的一台或多台机器上并行执行。

在一个Worker (进程) 中，分为多个Task (线程)，每个线程对应于Spout/Bolt的实现。

工作流程：

根据业务设计Topology

根据业务流程实现Spout的 nextTuple 接口中的数据输入

根据业务细节实现Bolt的 execute 接口中的处理逻辑

提交Topology开始执行

3.2.6.1 提交Topology时的参数

总Worker数目

if #worker <= 10 then
    _topology_master 以Task形式存在，不独占Worker
else
    _topology_master 以Task形式存在，独占Worker
end

每个component的并行度

并行度(parallelism) 代表有多少个Task线程来执行这个Spout/Bolt。

同一个Component中的Task id一定是连续的。

每个Component之间的关系

声明Spout和Bolt之间的对应关系，JStorm使用均匀调度算法，奇偶不同数目的Spout/Bolt会存在某个进程只有Spout或只有Bolt的情形。若topology运行过程中挂掉，JStorm会不断尝试重启进程。

3.2.7 消息通信

Spout发消息

JStorm 计算消息目标 Task Id列表

   if Task_id 在本进程 then
     直接将消息放入目标Task执行队列
   else
     netty跨进程发送至目标Task中
   end

3.2.8 实时计算结果输出

JStorm的Spout或Bolt中会有一个定时往外部存储写计算结果的逻辑，将数据按照业务需求被实时或近似实时地输出。

3.2.9 小结

JStorm是阿里巴巴平台的产品，相对来说适用于大量数据集群的情况，目前我现有的资源很难使用。因此，选择Python系的streamparser进行阅读。

3.3 折腾Storm平台部署

3.3.1 部署storm平台

下载[Java 8/9][1]、maven、zookeeper、storm、[lein][2]的release并依次安装。(以上库除lein外为storm运行所必须，由于服务器在国外，下载时间较长)

将 JDK、maven、zookeeper、storm 等拷贝至/opt 目录下，在~/.bash_profile中将相应目录加入PATH:

export JAVA_HOME="/opt/jdk8"
export MAVEN_HOME="/opt/maven"
export ZOO_KEEPER_HOME="/opt/zookeeper"
export STORM_HOME="/opt/storm"
PATH=$STORM_HOME/bin:$ZOO_KEEPER_HOME/bin:$MAVEN_HOME/bin:JAVA_HOME/bin:$PATH
export PATH

- 进入/opt/zookeeper/conf目录，编辑zoo.cfg配置文件，如下：

tickTime=2000
initLimit=10
syncLimit=5
dataDir=/var/zookeeper # 注意需要对该目录有写权限
clientPort=2181

进入/opt/storm/conf目录，编辑storm.yaml配置文件，如下：

storm.zookeeper.servers: # 注意此处有空格
  - "10.211.55.37"  # 填入配置机器的IP，若为集群则在下一行以同样格式列出
  # - "other server ip"
  
# 此处为Nimbus服务器地址，单机运行时无效，系统自动使用本地hostname，[原因待求证]
# nimbus.seeds:["host1","host2","host3"] 

storm.local.dir: "/var/storm" # 需要保证该目录有写权限，此处使用root账户所以不考虑。

# 设置supervisor slots
supervisor.slots.ports: # 注意此处有空格
  - 6700
  - 6701
  - 6702
  - 6702
# 此处在storm 1.1.1版的配置模板文件中未提及，但配置后在集群中能看到，[原因待求证]

启动zookeeper集群
```
bin/zkServer.sh start
```

在[Master服务器][3]上启动storm nimbus服务
```
bin/storm nimbus >> /dev/null &
```
在[Worker服务器][3]上启动storm supervisor服务
```
bin/storm supervisor >> /dev/null &
```
在[Master服务器][3]上启动storm UI工具
```
bin/storm ui &
```
在[Master服务器][3]上采用jps查看服务的启动情况，若显示config_value则表示服务正在初始化；若显示nimbus、supervisor、core、jps、QuorumPeerMain则说明初始化完毕，打开浏览器输入http://server_host:8080即可进入Storm UI查看相关信息。

[1] Java8/9 推荐安装Oracle官网下载的完整版JDK，因为后续的 lein 需要完整的JDK。解压JDK之后配置系统变量即可。(本次Linux机器采用 Java 8)

[2] lein全称为leiningen，是自动化管理Clojure脚本的工具，类似于Cargo。lein目前的脚本下载会出现证书不匹配的问题，解决方案为export HTTP_CLIENT="wget --no-check-certificate -O"。而且，上述设置后，下载release依旧很慢、需要代理，可以直接wget下载对应的release，放到~/.lein/self-installs/leiningen-2.5.3-standalone.jar即可，参考这里。lein是一个可执行脚本，需要放到/usr/bin或者/usr/local/bin下面，然后命令行中运行./lein和lein repl完成安装。

[3] 本地测试则仅仅在本机即可

3.3.2 案例工程 WordCount

主要参照《Get Started with Storm》一书，网上有中文版，此处参照为英文原版。

3.3.2.1 前提准备

maven编译工具，建立pom.xml来声明该工程的编译结构，包括注明编译需要的maven版本、编译所需的storm依赖库在线地址、以及依赖的storm版本。

  <repositories>
        <!-- Repository where we can found the storm dependencies  -->
        <repository>
            <id>clojars.org</id>
            <url>http://clojars.org/repo</url>
        </repository>
  </repositories>

3.3.2.2 编写对应代码文件

1. 建立文件结构

建立对应的文件结构src/main/java/{spouts,bolts}、/src/main/resources等，其中resources文件夹要存放相应的资源文件。

2. 编写`spouts`实例

package spouts;
import ....;

public class WordReader implements IRichSpout {
    private .....;
     public void ack(Object msgId) {...;}
     public void fail(Object msgId) {...;}
     public void nextTuple() {...;}
  // first method called in ANY spout    
  public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {...;}
     public void close() {}
     public void declareOutputFields(OutputFieldsDeclarer declarer) {...;}
}

3. 编写`bolts`实例

package bolts;
import ...;

public class WordNormalizer implements IRichBolt {
    private ...;
     public void cleanup(){}
     public void execute(Tuple input) {...;}
     public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {...;}
     public void declareOutputFields(OutputFieldsDeclarer declarer) {...;}
}

4. 编写`topology`结构

import ...;

public class TopologyMain {
    public static void main(String[] args) throws InterruptedException {
        // Topology definition
         TopologyBuilder builder = new TopologyBuilder();
         builder.setSpout();
         builder.setBolt().shuffleGrouping();
         builder.setBolt().fieldGrouping();
         // Configuration
         Config conf = new Config();
         conf.put("xxx", args[0]);
         conf.setDebug(false);
         // Topology run
         conf.put(Config.TOPOLOGY_MAX_SPOUT_PENDING, 1);
         LocalCluster cluster = new LocalCluster();
         cluster.submitTopology("xxxx",conf, builder.createTopology());
         Thread.sleep(1000); // sleep to reduce server load
         cluster.shutdown();
    }
}

5. 使用`mvm`带好相应参数运行

mvn clean install # maven会自动下载相关的包
cd target # 注意目录下有 `pom.xml` 中标识的输出的jar包
storm jar output-jar.jar path.to.your.topology # LocalCluster 执行，然后关闭
storm jar output-jar.jar path.to.your.topology name-of-storm # 提交jar至storm集群，循环执行，可在UI中查看

3.4 运行Storm例子程序的问题记录

3.4.1 存在问题以及解答记录

1. 程序中的`collector`是指的什么？

collector是用来追踪处理逻辑上每个emit的数据是否在下游bolt中被成功处理。collector是与storm通信的工具，反馈每个任务的处理情况。

2. 程序中`collector`最后`emit`的`Value(…)`是什么结构？

官方文档解释：A convenience class for making tuple values using new Values("field1", 2, 3) syntax.

Value是构建Tuple的一个元组类，该类实现了Serializable, Cloneable, Iterable<Object>, Collection<Object>, List<Object>, RandomAccess等接口，与Bolt中的execute接口相对应：public void execute(Tuple input, BasicOutputCollector collector)

3. Storm中的Ack/Fail机制中对`fail`情形的处理？

为了保证数据能正确的被处理, 对于spout产生的每一个tuple, storm都会进行跟踪。

这里面涉及到ack/fail的处理，如果一个tuple处理成功是指这个Tuple以及这个Tuple产生的所有Tuple都被成功处理, 会调用spout的ack方法；

如果失败是指这个Tuple或这个Tuple产生的所有Tuple中的某一个tuple处理失败, 则会调用spout的fail方法；

在处理tuple的每一个bolt都会通过OutputCollector来告知storm, 当前bolt处理是否成功。

另外需要注意的，当spout触发fail动作时，不会自动重发失败的tuple，需要我们在spout中重新获取发送失败数据，手动重新再发送一次。

4. Storm中的Ack原理

Storm中有个特殊的task名叫acker，他们负责跟踪spout发出的每一个Tuple的Tuple树（因为一个tuple通过spout发出了，经过每一个bolt处理后，会生成一个新的tuple发送出去）。当acker（框架自启动的task）发现一个Tuple树已经处理完成了，它会发送一个消息给产生这个Tuple的那个task。
Acker的跟踪算法是Storm的主要突破之一，对任意大的一个Tuple树，它只需要恒定的20字节就可以进行跟踪。
Acker跟踪算法的原理：acker对于每个spout-tuple保存一个ack-val的校验值，它的初始值是0，然后每发射一个Tuple或Ack一个Tuple时，这个Tuple的id就要跟这个校验值异或一下，并且把得到的值更新为ack-val的新值。那么假设每个发射出去的Tuple都被ack了，那么最后ack-val的值就一定是0。Acker就根据ack-val是否为0来判断是否完全处理，如果为0则认为已完全处理。

要实现ack机制：

spout发射tuple的时候指定messageId
spout要重写BaseRichSpout的fail和ack方法
spout对发射的tuple进行缓存(否则spout的fail方法收到acker发来的messsageId，spout也无法获取到发送失败的数据进行重发)，看看系统提供的接口，只有msgId这个参数，这里的设计不合理，其实在系统里是有cache整个msg的，只给用户一个messageid，用户如何取得原来的msg貌似需要自己cache，然后用这个msgId去查询，太坑爹了
spout根据messageId对于ack的tuple则从缓存队列中删除，对于fail的tuple可以选择重发。
设置acker数至少大于0；Config.setNumAckers(conf, ackerParal);

Storm的Bolt有BasicBolt和RichBolt:
　　在BasicBolt中，BasicOutputCollector在emit数据的时候，会自动和输入的tuple相关联，而在execute方法结束的时候那个输入tuple会被自动ack。
　　使用RichBolt需要在emit数据的时候，显式指定该数据的源tuple要加上第二个参数anchor tuple，以保持tracker链路，即collector.emit(oldTuple, newTuple)；并且需要在execute执行成功后调用OutputCollector.ack(tuple)，当失败处理时，执行OutputCollector.fail(tuple)。

由一个tuple产生一个新的tuple称为：anchoring，你发射一个tuple的同时也就完成了一次anchoring。

ack机制即，spout发送的每一条消息，在规定的时间内，spout收到Acker的ack响应，即认为该tuple被后续bolt成功处理；在规定的时间内（默认是30秒），没有收到Acker的ack响应tuple，就触发fail动作，即认为该tuple处理失败，timeout时间可以通过Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS来设定；或者收到Acker发送的fail响应tuple，也认为失败，触发fail动作

注意，如果继承BaseBasicBolt那么程序抛出异常，程序直接异常停止了，不会让spout进行重发。

5. Fail注意点小结

若某个task节点处理的tuple一直失败，会导致spout节点存储的tuple数据越来越多，直至内存溢出
在某个tuple的众多子tuple中，若某一个子tuple处理fail，但是其他子tuple仍会执行。即当所有子tuple都执行数据存储操作，其中一个子tuple出现fail，即使整个处理是fail，但是成功的子tuple仍会执行而不会滚。
Tuple的追踪只要是spout开始，可以在任意层次bolt停止追踪并作出应答。acker的数量可以通过Acker task组件来设置。
一个Topology并不需要太多acker，除非storm吞吐量不正常。
若不需要保证可靠性，即不追踪tuple树的执行情况，则系统里的消息数量会减少一半。
关闭消息可靠性的三种方法：
- config.Topology_ACKERS=0
- Spout发送消息时不指定消息的msgId
- 在emit方法中不指定输入消息

6. `Anchoring` 锚定概念

拓扑是一个消息(Tuple)沿着一个或多个分支的树节点，每个节点将ack(tuple)或者fail(tuple)，这样当消息失败时Storm就会知道，并通知Spout重发消息。因为一个Storm拓扑运行在一个高度并发的环境中，跟踪原始Spout示例的最好办法是在消息Tuple中包含一个原始Spout的引用，这种行为(技术)被称为Anchoring(锚定)。

锚点发生的语句在collector.emit(tuple, new Values(word))中，传递元组(emit方法)使Storm能够跟踪原始Spout。collector.ack(tuple)和collector.fail(tuple)告诉Spout知道每个消息的处理结果。当消息树上的每个消息已经被处理，Storm认为来自Spout的元组被完全处理。当消息树在一个可配置的超时内处理失败，一个元组被认为是失败的。处理的每一个元组必须是确认或者失败，Storm会使用内存来追踪每个元组，如果不对每个元组进行确认/失败，最终会耗尽内存。

为了简化编码，Storm为Bolt提供了一个IBasicBolt接口，它会在调用execute()方法之后正确调用ack()方法，BaseBasicBolt类是该接口的一个实现，用来实现自动确认。

7. Storm组件与编程时遇到的概念

名称	解释
Nimbus	负责资源分配和任务调度，Nimbus分配任务到Zookeeper指定目录。
Supervisor	去Zookeeper指定目录接受Nimbusf分配的任务，启停自己的Worker进程。
Worker	运行具体处理组件逻辑的进程(process)，Worker的任务分为Spout和Bolt两种。
Task	Worker启动相应的物理线程(Executor)，Worker执行的每一个Spout/Bolt线程成为一个Task，0.8版本后Spout/Bolt的Task可能共享一个Executor。
Topology	拓扑，Storm集群，即定义的数据流处理的DAG。
Spout	Storm集群的数据源
Bolt	Storm任务的处理逻辑单元，在集群多个机器上并发执行。
Tuple	消息元组，Spout、Bolt用来与Storm集群通信、反馈任务处理成功与否的载体。恒定为20Bit。
Stream groupings	数据流的分组策略，分7种，常见为`shuffleGrouping()`、`fieldsGrouping()`。
Executor	Worker启动的实际物理线程，一般一个Executor执行一个Task，但也能执行多个Task。
Configuration	Topology的配置

8. 序列化与反序列化

由于博客上特别提到了Java虚拟机序列化的性能极其辣鸡，所以在此记录。

把对象转换为字节序列的过程称为对象的序列化；把字节序列恢复为对象的过程称为对象的反序列化。

对象的序列化主要有两种用途：

把对象的字节序列永久地保存到硬盘上，通常存放在一个文件中；

在网络上传送对象的字节序列。

在很多应用中，需要对某些对象进行序列化，让它们离开内存空间，入住物理硬盘，以便长期保存。比如最常见的是Web服务器中的Session对象，当有 10万用户并发访问，就有可能出现10万个Session对象，内存可能吃不消，于是Web容器就会把一些seesion先序列化到硬盘中，等要用了，再把保存在硬盘中的对象还原到内存中。

当两个进程在进行远程通信时，彼此可以发送各种类型的数据。无论是何种类型的数据，都会以二进制序列的形式在网络上传送。发送方需要把这个Java对象转换为字节序列，才能在网络上传送；接收方则需要把字节序列再恢复为Java对象。

在Java/Storm中，可以理解为toString()函数的自定义实现。注意使用transient 修饰的对象无法序列化。

9. `declareOutputFields()` 函数的具体作用

该Spout代码里面最核心的部分有两个：

用collector.emit()方法发射tuple。我们不用自己实现tuple，我们只需要定义tuple的value，Storm会帮我们生成tuple。Values对象接受变长参数。Tuple中以List存放Values，List的Index按照new Values(obj1, obj2,…)的参数的index，例如我们emit(new Values("v1", "v2"))，那么Tuple的属性即为：{ [ "v1" ], [ "V2" ] }
declarer.declare方法用来给我们发射的value在整个Stream中定义一个别名。可以理解为key。该值必须在整个topology定义中唯一。

3.5 Windows 平台部署本地测试环境的注意事项

3.5.1 所需安装包

Java SE Development Kit 8/9，安装到非C:\Program Files\目录下，否则storm将无法启动。
Apache-maven，解压到本地目录，推荐非系统盘
Zookeeper，解压到本地目录，推荐非系统盘
Storm，解压到本地目录，推荐非系统盘

3.5.2 环境变量配置

JAVA_HOME: path\to\your\jdk-file
Path: path\to\your-storm\bin;path\to\your-zookeeper\bin;path\to\your-maven\bin;%JAVA_HOME%\bin

3.5.3 配置文件设定

配置zoo.cfg，参照3.1
配置storm.yaml，参照3.1，注意storm.local.dirs中的目录使用\\来表示\。

3.5.6 启动集群

zkServer
storm nimbus
storm supervisor
storm ui
# 打开浏览器 http://127.0.0.1:8080/index.html