JStorm学习

Posted aibabel

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了JStorm学习相关的知识,希望对你有一定的参考价值。

一、简介

     JStorm是一个分布式实时计算引擎。JStorm是一个类似于Hadoop MapReduce的系统,用户按照指定的接口实现一个任务,然后将这个任务交给JStorm系统,JStorm将这个任务跑起来,并按7*24小时运行。如果中间一个worker发生了意外故障,调度器立即分配一个新的worker来替换这个失效的worker。 从应用的角度上看,JStorm是一种遵循某种编程规范的分布式应用;从系统的角度上看,JStorm是一套类似MapReduce的调度系统;从数据角度上看,JStorm是一套基于流水线的消息处理机制。 JStorm通过一系列基本元素实现实时计算的目标,其中包括topology,spout,bolt等。JStorm在模型上和MapReduce有很多相似的地方。下表是JStorm组件和MapReduce组件的对比:   JStorm MapReduce 角色 Nimbus JobTracker Supervisor TaskTracker Worker Child 应用名称 Topology Job 编程接口 Spout/Bolt Mapper/Reduce 实时计算任务需要打包成Topology提交,和MapReduce Job相似,不同的是,MapReduce Job在计算完后就结束,而JStorm的Topology任务一旦提交,就永远不会结束,除非显示停止。

二、JStorm系统架构

     JStorm的系统架构如下所示: JStorm与Hadoop相似,保持了Master/Slaves简洁优雅的架构。与Hadoop不同的是,JStorm的Master/Salves之间不能直接通过RPC来交换心跳信息,而是借助Zookeeper来实现。这样的设计虽然引入了第三方依赖,但是简化了Nimbus/Supervisor的设计,同时也极大提高了系统的容错能力。 整个JStorm系统中共存三类不同的daemon进程,分别为Nimbus,Supervisor和Worker。 Nimbus:JStorm的主控节点,作为调度器的角色。 Nimbus类似于MapReduce的JobTracker,负责接收和验证客户端提交的topology;分配任务;向ZK写入任务相关的元信息。此外,Nimbus还负责通过ZK来监控节点和任务的健康情况。当有Supervisor节点变化和Worker进程出现问题时及时进行任务重新分配。Nimbus分配任务的结果不是直接下发到Supervisor,而是通过ZK维护分配数据进行过渡。 特别地,JStorm 0.9.0领先Apache Storm实现了Nimbus HA,由于Nimbus是Stateless节点,所有的状态信息都交由ZK托管,所以HA相对比较简单,热备Nimbus subscribe ZK关于Master活跃状态数据,一旦发现Master出现问题即从ZK里恢复数据后可以立即接管。 从0.9.0开始,JStorm提供非常强大的调度功能,基本上可以满足大部分的需求,同时支持自定义任务调度策略。JStorm的资源不再仅是Worker的端口,而从CPU/Memory/Disk/Port等四个维度综合考虑。

Nimbus任务调度算法如下:

0)优先使用自定义任务分配算法,当资源无法满足需求时,该任务放到下一级任务分配算法;

1)使用历史任务分配算法(如果打开使用历史任务属性),当资源无法满足需求时,该任务放到下一级任务分配算法;

2)使用默认资源平衡算法,计算每个Supervisor上剩余资源权值,取权值最高的Supervisor分配任务。 Supervisor:JStorm的工作节点,作为Worker的代理角色,负责杀死worker或运行worker。 Supervisor类似于MapReduce的TaskTracker,subscribe ZK分配到该节点的任务数据。Supervisor根据Nimbus的任务分配情况来启动/停止工作进程Worker。Supervisor需要定期向ZK写入活跃端口信息以便Nimbus及时监控。Supervisor不执行具体的数据处理任务,所有的数据处理任务都交给Worker。

Worker:JStorm中的任务执行者,是Task的容器。Worker类似于MapReduce的Task,所有实际的数据处理工作都在worker内执行完成。Worker需要定期向Supervisor汇报心跳。由于在同一个节点,同时为保持节点的无状态,Worker定期将状态信息写入到本地磁盘。

Supervisor通过读取本地磁盘状态信息完成心跳交互过程,Worker绑定一个独立端口,Worker内所有单元共享Worker的通信能力。 完整的Topology任务是由分布在多个Supervisor节点上的Worker进程(JVM)来执行,每个Worker都执行且仅执行Topology任务的一个子集。

Task:真正任务的执行者 执行具体数据处理实体,也就是用户实现的Spout/Blot实例。

ZK:是整个系统的协调者 Nimbus、Supervisor和Worker均为Stateless节点,支持Fail-Fast,这为JStorm的扩展性和容错能力提供了很好的保障。

三、JStorm基本概念

 

stream 在JStorm中有对于流stream的抽象,流是一个不间断的无界的连续tuple。在Topology中,spout是stream的源头,负责从特定数据源stream发射;bolt接收任意多个stream输入,然后进行数据加工处理。bolt还可以发射出新的stream到下游的bolt。 spout JStorm的消息源。JStorm认为每个stream都有一个stream源,也就是原始元组的源头,所以它将这个源头抽象为spout,spout可能是连接消息中间件(如MetaQ, Kafka, TBNotify等),并不断发出消息,也可能是从某个队列中不断读取队列元素并装配为tuple发射。 bolt JStorm的消息处理者。bolt用于为Topology进行消息处理,它可以执行查询、过滤、聚合及各种复杂运算操作。Bolt的消息处理结果可以作为下游Bolt的输入不断迭代。bolt可以消费任意数量的输入流,只要将流方向导向该bolt。 Tuple JStorm将流中数据抽象为tuple,存在于任意两个有数据交互的组件(Spout/Bolt)之间。一个tuple就是一个值列表          value list,list中的每个value都有一个name,并且该value可以是基本类型,字符类型,字节数组等,当然也可以是其他可序列化的类型。简单来说,Tuple就是一组实现了序列化器带有名称的Java对象集合。 拓扑的每个节点都要说明它所发射出的元组的字段的name,其他节点只需要订阅该name就可以接收处理。 topology 拓扑是JStorm中最高层次的一个抽象概念,它可以被提交到JStorm集群执行,一个拓扑就是一个数据流转换图。计算任务Topology是由不同的Spout和Bolt通过Stream连接起来的DAG图。典型的topology的结构示意图如下所示: 从整个Topology上看,Spout/Bolt可以看作DAG的节点,Stream是连接不同节点之间的有向边,Tuple则是流过Stream的数据集合。下面是一个Topology内部Spout和Bolt之间的数据流关系: Topology中每一个计算组件(Spout和Bolt)都有一个并行度,在创建Topology时指定(默认为1),JStorm在集群内分配对应个数的线程Task并行。 资源slot 在JStorm中,资源类型分为4种:CPU、Memory、Disk、Port,不再局限于Storm的port。 即一个supervisor可以提供多少个CPU slot,多少个Memory slot, 多少个Disk slot, 多少个Port slot。 一个worker就消耗一个Port slot, 默认一个task会消耗一个CPU slot和一个Memory slot 当task执行任务较重时,可以申请更多的CPU slot, 当task需要更多内存时,可以申请更多的内存slot, 当task 磁盘读写较多时,可以申请磁盘slot,则该磁盘slot给该task独享 四、数据流分发策略 spout/bolt都会有多个线程来并发执行,那么如何在两个组件(Spout和Bolt)之间发送Tuple呢?JStorm通过定义Topology时为每个bolt指定输入stream以及指定提供的若干种数据流分发(Stream Grouping)策略来解决这个问题。

JStorm提供了8种数据流分发策略: Shuffle Grouping:随机分组 随机派发stream里面的tuple,保证每个bolt接收到的tuple数目大致相同。通过轮询随机的方式,使得下游bolt之间接收到的tuple数目差值不超过1。 Fields Grouping:按字段分组 具有同样字段值的tuple会被分到相同的bolt里的Task,不同字段则会被分配到不同的Task中。 All Grouping:广播分组 每一个tuple都能被所有的bolt收到。 Global Grouping:全局分组 tuple被分配到bolt中id值最小的一个task中。 Non Grouping:不分组 tuple会按完全随机的方式分发到下游bolt。 Direct Grouping:直接分组 tuple需要指定由bolt的哪个task来接收。只有被声明为Direct Stream的消息流可以声明这种分组方法。 Local or Shuffle Grouping:基本同Shuffle Grouping Custom Grouping:用户自定义分组策略 CustomStreamGrouping是自定义分组策略时用户需要实现的接口。

五、计算组件映射到计算资源

   Topology的计算组件(spout/bolt)如何映射到计算资源上?首先先明确Worker/Executor/Task之间的关系: Worker 完整的Topology任务是由分布在Supervisor节点上的多个Worker进程(JVM)来执行的。每个Worker都执行且仅执行Topology任务的一个子集。 Executor Worker内部会有一个或多个Executor,每个Executor对应一个线程。Executor包括SpoutExecutor和BoltExecutor,同一个Worker里所有的Executor只能属于某一个Topology里的执行单元。 Task 执行具体数据的处理实体。也是用户实现的spout/bolt实例。一个Excutor对应多个Task,在定义Topology可以指定Task。默认Executor和Task一一对应。这就是说,系统中Executor数量一定是小于等于Task数量(#Executor≤#Task)。 下图给出了一个简单的例子: 上半部分描述的是Topology结构及相关说明,其中定义了整个Topology的worker=2、DAG关系、各个计算组件的并行度;下半部分描述了Topology的Task在Supervisor节点的分布情况。从中可以看出Topology到Executor之间的关系。 0、Worker数在提交Topology时在配置文件中指定; 例:#Worker=2 1、执行线程/Executor数在定义Topology的各计算组件并行度时决定,可以不指定,默认为1。其中各个计算组件的并行度之和即为该Topology执行线程总数。 例:#Executor=sum(#parallelism hint)=2+2+6=10 2、Task数目也在定义Toplogy时确定,若不指定默认每个Executor线程对应一个Task,若指定Task数目会在指定数目的线程里平均分配。 例:#Task=sum(#task)=2+4+6=12,其中Executor4={Task0,Task1} 六、Ack机制 Ack 机制是storm整个技术体系中非常闪亮的一个创新点, JStorm很好的继承了这个机制,并对原生storm的ack机制做了一点点代码优化。 通过Ack机制,spout发送出去的每一条消息,都可以确定是被成功处理或失败处理, 从而可以让开发者采取动作。比如在Meta中,成功被处理,即可更新偏移量,当失败时,重复发送数据。因此,通过Ack机制,很容易做到保证所有数据均被处理,一条都不漏。另外需要注意的,当spout触发fail动作时,不会自动重发失败的tuple,需要spout自己重新获取数据,手动重新再发送一次。 如图当定义Topology时指定Acker,JStorm除了Topology本身任务外,还会启动一组称为Acker的特殊任务,负责跟踪Topolgogy DAG中的每个消息。每当发现一个DAG被成功处理完成,Acker就向创建根消息的Spout任务发送一个Ack信号。Topology中Acker任务的并行度默认parallelism hint=1,当系统中有大量的消息时,应该适当提高Acker任务的并行度。 Acker按照Tuple Tree的方式跟踪消息。当Spout发送一个消息的时候,它就通知对应的Acker一个新的根消息产生了,这时Acker就会创建一个新的Tuple Tree。当Acker发现这棵树被完全处理之后,他就会通知对应的Spout任务。 Acker任务保存了数据结构Map<MessageID,Map< TaskID, Value>>,其中MessageID是Spout根消息ID,TaskID是Spout任务ID,Value表示一个64bit的长整型数字,是树中所有消息的随机ID的异或结果。通过TaskID,Acker知道当消息树处理完成后通知哪个Spout任务,通过MessageID,Acker知道属于Spout任务的哪个消息被成功处理完成。Value表示了整棵树的的状态,无论这棵树多大,只需要这个固定大小的数字就可以跟踪整棵树。当消息被创建和被应答的时候都会有相同的MessageID发送过来做异或。当Acker发现一棵树的Value值为0的时候,表明这棵树已经被成功处理完成。 例如,对于前面Topology中消息树,Acker数据的变化过程: Step0.A发送T0给B后: R0=r0 <id0,<taskA,R0>> Step1.B接收到T0并成功处理后向C发送T1,向D发送T2: R1=R0^r1^r2=r0^r1^r2 <id0,<taskA,R0^R1>> =<id0,<taskA,r0^r0^r1^r2>> =<id0,<taskA,r1^r2>> Step2.C接收到T1并成功处理后: R2=r1 <id0,<taskA,r1^r2^R2>> =<id0,<taskA,r1^r2^r1>> =<id0,<taskA,r2>> Step3.D接收到T2并成功处理后: R3=r2 <id0,<taskA,r2^R3>> =<id0,<taskA,r2^r2>> =<id0,<taskA,0>> 当结果为0时Acker可以通知taskA根消息id0的消息树已被成功处理完成。 需要指出的是,Acker并不是必须的,当实际业务可以容忍数据丢失情况下可以不用Acker,对数据丢失零容忍的业务必须打开Acker,另外当系统的消息规模较大是可适当增加Acker的并行度。                 七、故障恢复

           1)节点故障 Nimbus故障 Nimbus本身无状态,所以Nimbus故障不会影响正在正常运行任务,另外Nimbus HA保证Nimbus故障后可以及时被备份Nimbus接管。 Supervisors节点故障 Supervisor故障后,Nimbus会将故障节点上的任务迁移到其他可用节点上继续运行,但是Supervisor故障需要外部监控并及时手动重启。 Worker故障 Worker健康状况监控由Supervisor负责,当Woker出现故障时,Supervisor会及时在本机重试重启。 Zookeeper节点故障 Zookeeper本身具有很好的故障恢复机制,能保证至少半数以上节点在线就可正常运行,及时修复故障节点即可。

       2)任务失败 Spout失败 消息不能被及时被Pull到系统中,造成外部大量消息不能被及时处理,而外部大量计算资源空闲。 Bolt失败 消息不能被处理,Acker持有的所有与该Bolt相关的消息反馈值都不能回归到0,最后因为超时最终Spout的fail将被调用。 Acker失败 Acker持有的所有反馈信息不管成功与否都不能及时反馈到Spout,最后同样因为超时Spout的fail将被调用。 任务失败后,需要Nimbus及时监控到并重新分配失败任务。

 

八、关键流程

1、Topology提交 JStorm为用户提供了StormSubmitter. submitTopology用来向集群提交Topology,整个提交流程: Client端: 0)客户端简单验证;

1)检查是否已经存在同名Topology; 2)提交jar包; 3)向Nimbus提交Topology; Nimbus端: 0)Nimbus端简单合法性检查; 1)生成Topology Name; 2)序列化配置文件和Topology Code; 3)Nimbus本地准备运行时所需数据; 4)向ZK注册Topology和Task; 5)将Task压入分配队列等待TopologyAssign分配;

九、基本接口

1、IRichSpout IRichSpout 为最简单的Spout接口 IRichSpout{ @Override public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) { } @Override public void close() { } @Override public void activate() { } @Override public void deactivate() { } @Override public void nextTuple() { } @Override public void ack(Object msgId) { } @Override public void fail(Object msgId) { } @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { } @Override public Map<String, Object> getComponentConfiguration() { return null; } 其中注意: spout对象必须是继承Serializable, 因此要求spout内所有数据结构必须是可序列化的 spout可以有构造函数,但构造函数只执行一次,是在提交任务时,创建spout对象,因此在task分配到具体worker之前的初始化工作可以在此处完成,一旦完成,初始化的内容将携带到每一个task内(因为提交任务时将spout序列化到文件中去,在worker起来时再将spout从文件中反序列化出来)。 open是当task起来后执行的初始化动作 close是当task被shutdown后执行的动作 activate 是当task被激活时,触发的动作 deactivate 是task被deactive时,触发的动作 nextTuple 是spout实现核心, nextuple完成自己的逻辑,即每一次取消息后,用collector 将消息emit出去。 ack, 当spout收到一条ack消息时,触发的动作 fail, 当spout收到一条fail消息时,触发的动作 declareOutputFields, 定义spout发送数据,每个字段的含义 getComponentConfiguration 获取本spout的component 配置

2、IRichBolt IRichBolt 为最简单的Bolt接口 IRichBolt { @Override public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) { } @Override public void execute(Tuple input) { } @Override public void cleanup() { } @Override public void declareOutputFields(OutputFieldsDeclarer declarer) { } @Override public Map<String, Object> getComponentConfiguration() { return null; } } 其中注意: bolt对象必须是继承Serializable, 因此要求spout内所有数据结构必须是可序列化的 bolt可以有构造函数,但构造函数只执行一次,是在提交任务时,创建bolt对象,因此在task分配到具体worker之前的初始化工作可以在此处完成,一旦完成,初始化的内容将携带到每一个task内(因为提交任务时将bolt序列化到文件中去,在worker起来时再将bolt从文件中反序列化出来)。 prepare是当task起来后执行的初始化动作 cleanup是当task被shutdown后执行的动作 execute是bolt实现核心, 完成自己的逻辑,即接受每一次取消息后,处理完,有可能用collector 将产生的新消息emit出去。 在executor中,当程序处理一条消息时,需要执行collector.ack ;在executor中,当程序无法处理一条消息时或出错时,需要执行collector.fail  declareOutputFields, 定义bolt发送数据,每个字段的含义 getComponentConfiguration 获取本bolt的component 配置

以上是关于JStorm学习的主要内容,如果未能解决你的问题,请参考以下文章

Flink 在 字节跳动

JStorm介绍

JStorm:任务调度

jstorm

Jstorm

jstorm集成kafka