大数据-Storm流式框架(四)---storm容错机制

举报
前进的蜗牛 发表于 2023/12/18 09:44:53 2023/12/18
【摘要】 ​1、集群节点宕机Nimbus服务器     硬件单点故障?可以搭建HA jStorm搭建  nimbus的HAnimbus的信息存储到zookeeper中,只要下游没问题(进程退出)nimbus退出就不会有问题,如果在nimbus宕机,也不能提交作业。非Nimbus服务器     硬件supervisor故障时,该节点上所有Task任务都会超时,Nimbus会将这些Task任务重新分配到其...


1、集群节点宕机


Nimbus服务器     硬件

单点故障?可以搭建HA jStorm搭建  nimbus的HA

nimbus的信息存储到zookeeper中,只要下游没问题(进程退出)nimbus退出就不会有问题,

如果在nimbus宕机,也不能提交作业。

非Nimbus服务器     硬件

supervisor故障时,该节点上所有Task任务都会超时,Nimbus会将这些Task任务重新分配到其他服务器上运行

2、进程挂掉

Worker

挂掉时,Supervisor会重新启动这个进程。如果启动过程中仍然一直失败,并且无法向zookeeper发送心跳,Nimbus会将该Worker执行的任务重新分配到其他服务器上

Supervisor

无状态(所有的状态信息都存放在Zookeeper中来管理)

快速失败(每当遇到任何异常情况,都会自动毁灭)

Nimbus

无状态(所有的状态信息都存放在Zookeeper中来管理)

快速失败(每当遇到任何异常情况,都会自动毁灭)

3、消息的完整性

从Spout中发出的Tuple,以及基于他所产生Tuple(例如上个例子当中Spout发出的句子,以及句子当中单词的tuple等)。由这些消息就构成了一棵tuple树。当这棵tuple树发送完成,并且树当中每一条消息都被正确处理,就表明spout发送消息被“完整处理”,即消息的完整性。

元组树

如果衍生元组出错,则重发根元组

根元组就是spout发送的那个元组


Acker -- 消息完整性的实现机制

Storm的拓扑当中特殊的一些任务

负责跟踪每个Spout发出的Tuple的DAG(有向无环图)

默认每个worker一个acker

保证消息至少处理一次

Storm提供了几种不同级别的保证消息处理,包括尽力而为(best effort),至少一次(at least once),以及通过Trident保证只完全处理一次(exactly once)。

此处指的是至少完全处理一次

当元组树已经用完并且树中的每条消息都已处理完毕时,Storm会认为从水龙头发射的元组是“完全处理”的。如果在指定的超时内无法完全处理其消息树,则认为该元组失败。可以使用Config.TOPOLOGY_MESSAGE_TIMEOUT_SECS对指定的拓扑进行配置,默认为30秒

public interface ISpout extends Serializable {
    void open(Map conf, TopologyContext context, SpoutOutputCollector collector);
    void close();
    void nextTuple();
    void ack(Object msgId);
    void fail(Object msgId);
}

首先,storm调用spout的nextTuple方法发射一个元组。spout使用SpoutOutputCollector在声明的流中发射元组。当发射元组的时候,spout使用messageId标记该元组。

_collector.emit(new Values("field1", "field2", 3) , msgId);

其次,当向消费闪电发送元组的时候,strom追踪该消息树。如果storm发现一个元组被完全处理了,storm就会调用spout的ack方法并将该元组的messageId传送给它。如果元组处理超时,storm就调用spout的fail方法。只会在发射该元组的spout中调用它的ack或者fail方法。


好处是:

首先,当创建了新的元组树边的时候,要通知storm。其次,当完成了一个元组的处理之后要告诉storm。通过这种方式,storm就可以知道元组被完全处理了,然后调用ack方法,或者调用fail方法,如果处理失败。

在元组树中创建一条边,称为锚点。当发送一个新元组的时候就会创建一条边。

public class SplitSentence extends BaseRichBolt {
        OutputCollector _collector;

        public void prepare(Map conf, TopologyContext context, OutputCollector collector) {
            _collector = collector;
        }

        public void execute(Tuple tuple) {
            String sentence = tuple.getString(0);
            for(String word: sentence.split(" ")) {
                _collector.emit(tuple, new Values(word));
            }
            _collector.ack(tuple);
        }

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

每个元组通过在emit中指定输入的元组进行锚点标记

由于被标记锚点,如果在下游处理中处理失败了,元组树顶点的元组会重新发送。

_collector.emit(new Values(word));

上述方式没有对元组进行锚点标记。当元组在下游处理失败了,根元组不会重发。输出的元组也可以标记多个锚点。如果流中有聚合或者join,就比较有用。标记了多个锚点的元组如果处理失败,就会触发多个元组树根元组重发。通过list集合为元组标记多个锚点:

List<Tuple> anchors = new ArrayList<Tuple>();
anchors.add(tuple1);
anchors.add(tuple2);
_collector.emit(anchors, new Values(1, 2, 3));

很明显,手动调用fail方法比通过元组的处理超时从而由storm调用fail方法要更快地重发根元组。由于storm使用内存来存储元组树,如果不及时的ack或者fail有可能导致内存溢出。


storm的拓扑中提供了一组acker用于追踪spout发射的每个元组及其衍生的元组,一旦发现DAG处理完了,就同创建该元组的spout进行确认。

Config.TOPOLOGY_ACKERS用于设置acker的数量。默认情况下一个worker一个acker任务

当拓扑创建了元组,就会为其分配一个随机的64bit的id,acker使用该ID追踪spout发送的每个元组。元组树上的元组都知道这个ID。当在闪电中创建了新的元组,该ids会拷贝给新的元组。当元组确认后,元素发送消息给acker任务,以改变元组树。


当通过C衍生出D和E并且确认之后,就会从元组树中移除C。这样可以保证元组树不会过早地完成。

如果拓扑中包含多个acker,当一个元组确认后,如何知道向哪个acker发送确认消息?

storm使用hash取模的方式将一个spout的元组id跟一个acker任务绑定。

acker任务如何知道该向哪个spout任务发送确认消息?spout发射元组的时候会给合适的acker发送一个消息表示对哪个spout的元组负责。acker发现元组树完成了,就知道向哪个spout任务发送完成的消息。

acker不显式地追踪元组。如果有数十万的节点,追踪所有的元组会有耗尽acker内存的风险。acker使用一个定长的空间20字节做这个工作。这个是storm的主要创新。

acker将spout发射的元组id和一个64bit的数字ack val)相关联。ack val代表了该元组树的状态,不管spout发射的元组及其衍生的元组有多少,它仅仅是对所有创建的元组以及确认的元组id求异或xor操作如果最终这个值ack val变成0,表示元组树已经被完全处理。某个元组的id和该64bit的数字异或结果是0的情况极其少见,比如每秒处理10k的元组,需要5000万年才会产生一个错误,造成数据丢失。

元组都会产生和消亡,元组不会凭空产生,也不会凭空消失

元组守恒定理

acker为当前根元组预留一个000二进制数字

messageId

000

100

100

010

110

001

111

100

011

010

001

001

000

如果这个值不是0,则过30s,要求spout重发根元组。

acker

异常情况:

  1. 由于任务死掉,没有确认元组:元组树根节点的元组和丢掉的元组确认会超时,重新发送根元组。
  2. acker死掉:所有的元组确认都会超时,根节点元组重发
  3. spout死掉:spout获取数据的数据源负责重新发送消息。例如:MQ等会将所有打开的消息放回到队列中,之后重新处理。


由此可见,strom的可靠性机制完全是分布式的,可扩展的以及容错的。

【版权声明】本文为华为云社区用户原创内容,转载时必须标注文章的来源(华为云社区)、文章链接、文章作者等基本信息, 否则作者和本社区有权追究责任。如果您发现本社区中有涉嫌抄袭的内容,欢迎发送邮件进行举报,并提供相关证据,一经查实,本社区将立刻删除涉嫌侵权内容,举报邮箱: cloudbbs@huaweicloud.com
  • 点赞
  • 收藏
  • 关注作者

评论(0

0/1000
抱歉,系统识别当前为高风险访问,暂不支持该操作

全部回复

上滑加载中

设置昵称

在此一键设置昵称,即可参与社区互动!

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。

*长度不超过10个汉字或20个英文字符,设置后3个月内不可修改。