标签:
KafkaSpout的核心逻辑都是由PartitionManager来实现的。
但是这个类实现时候需要考虑的东西有些多,0.92至0.93,至当前(2015.3.14)的master一直在变化。在这里,先分析一下最近的发布版0.93里的逻辑。也提出一些问题,希望以后Apache Storm会把这个类实现地更完美一些。
PartitionManager的主要功能
PartitionManager用来管理单个Partition。提供持久化读取进度、读取消息功能,并提供Storm的spout需要实现的nextTuple, fail, ack等功能。
实现PartitionManager需要考虑的问题
有一些问题是设计PartitionManager时必须考虑的,先把他们提一下,然后看下0.93版PartitionManager的实现。
关于批量读取消息以及缓存消息
由于Kafka的实现细节(为了高吞吐而设计的log格式,通讯协议),Kafka的SimpleConsumer每次读取消息是会去读取一批,而不能指定响应想要包含的具体的offset,并且由于消息按批压缩,使得每次读取的响应包含的offset可能比想要的最小的offset还要小(但不会更大)。所以,对于PartitoinManager来说,在内部构造一个缓存,保存每次读取得到的一批message是一种自然而且高效的方式。
允许有超过一个message处于pendding(已发送但没有ack)状态?
如果在发射一个message的tuple之后,就开始等待。那么ack、fail、commit的逻辑就会很简单。但这样消息的处理效率会被极大的降低,且不说还可能使得下游bolt的一些task没事可做。所以一定得允许多个message正在被blot处理,也就是需要有pendding messages的集合。
有了pendding的messages集合,ack, fail, commit的逻辑就变得比较复杂,且需要做出一些折衷。
- 当有message对应的tuple失败时,如何在处理其它正常的消息时,特殊处理失败的这些message?
- 如果有message产生的tuple在多次重复后仍然失败,应该怎么做?丢弃它吗?
- 在Zookeeper中应该记录什么信息?
- 如果下游的bolt处理的进度太慢怎么办?如何衡量处理速度是否达到需求?
PartitionManager的具体实现
在Zookeeper中记录信息的意义:
下面是PartitionManager的commit方法的主要部分:
Map<Object, Object> data = (Map<Object, Object>) ImmutableMap.builder() .put("topology", ImmutableMap.of("id", _topologyInstanceId, "name", _stormConf.get(Config.TOPOLOGY_NAME))) .put("offset", lastCompletedOffset) .put("partition", _partition.partition) .put("broker", ImmutableMap.of("host", _partition.host.host, "port", _partition.host.port)) .put("topic", _spoutConfig.topic).build(); _state.writeJSON(committedPath(), data);
_committedTo = lastCompletedOffset;
- topology.id 记录了这个topology实例的id。当PartitionManager的构造函数被调用时,它会从Zookeeper里获取topology.id,以判断当前的task是否跟记录zookeeper里信息的是一个topology实例,如果不是,说明这是一个新提交的topology,这时,会判断是否设置了forceFromStart,如果是同一个topology实例,就不理会forceFromStart
- topology.name topology的名字,这个目前没用到。
- offset 在这个offset之前(不包括它)的所有消息都已经处理完成。
- partition partition id
- broker 此partition的leader
- topic partition所属的topic。注意,在PartitionManager初始化时,它并没有判断这个spout task的topic跟记录里的一致。所以,如果两个topology, 有同样的SpoutConfig.id,但是不同的topic,会引发混乱。
另外,这个JSON数据写的路径committedPath也是很重要的。PartitionManager初始化时,会从这个committedPath路径读取信息。
private String committedPath() { return _spoutConfig.zkRoot + "/" + _spoutConfig.id + "/" + _partition.getId(); }
所以,如果spoutConfig.id配置得不当,KafkaSpout就无法获取正确的进度记录。
另外,在所有记录里,最重要的就是offset这个记录。它的意义,使得PartitionManager不得不做出很多权衡。
PartitionManager用到的集合和记录
Long _emittedToOffset; SortedSet<Long> _pending = new TreeSet<Long>(); SortedSet<Long> failed = new TreeSet<Long>(); Long _committedTo; LinkedList<MessageAndRealOffset> _waitingToEmit = new LinkedList<MessageAndRealOffset>(); long numberFailed, numberAcked;
- _pending 所有己读取,但还没有被ack的消息,都在这里。
- failed 所有己经认定为failed的tuple来自的message的offset都在这里
- _waitingToEmit 所有己经被读取,但是还没经过“解析,emit tuple"步骤的消息都在这。
- _emittedToOffset offset小于它的消息都已经被读取了
- _comittedTo 所有offset小于它的消息都已被ack,或者由于某些原因不再需要被处理。
当PartitionManager的next方法被调用以emit新tuple时,它只会从_waitingToEmit取消息。那么failed里的消息如何被再重试呢?原因在于_waitingToEmit为空时,next方法会调用fill方法,而fill方法会考虑到failed集合内的元素,不过是一种比较粗放的做法。
fill方法
fill方法的主要逻辑依次分为三个部分:
- 判断该从哪个offset开始,从Kafka抓取消息
- 抓取消息,处理offset out of range 异常
- 把抓取到的消息放到_waitingToEmit集合中,同时与failed集合与pendding集合交互。
第一部分:
final boolean had_failed = !failed.isEmpty(); // Are there failed tuples? If so, fetch those first. if (had_failed) { offset = failed.first(); } else { offset = _emittedToOffset; }
这段代码里,offset即是将要从Kafka里抓取消息的offset。当failed集合不为空时,就用failed集合的最小的offset做为下次要抓取的offset。Kafka的FetchRequest每次会从Kafka中获取一批消息。所以,如果有消息fail,而此failed消息之后的消息已被ack,那么fill方法会重新获取这些已被ack的消息,从而使得这部分消息最终被重复处理。
如果没有failed消息,fill方法就会从之前读取过的最大的offset继续抓取。在知道了从何处抓取之后,开始真正的抓取过程:
try { msgs = KafkaUtils.fetchMessages(_spoutConfig, _consumer, _partition, offset); } catch (UpdateOffsetException e) { //如果是offset "out of range", 并且设置了useStartOffsetTimeIfOffsetOutOfRange _emittedToOffset = KafkaUtils.getOffset(_consumer, _spoutConfig.topic, _partition.partition, _spoutConfig); LOG.warn("Using new offset: {}", _emittedToOffset); // fetch failed, so don‘t update the metrics return; }
出现了UpdateOffsetException代表出现了这种情况:想要抓取的offset不在Kafka能提供的offset所在的范围之内,并且已经在config里设置了useStartOffsetTimeIfOffsetOutOfRange为true。想要抓取的offset不在Kafka提供的范围可能有几种原因:这部分消息被Kafka的log retention功能给删除了;leader变更,使得部分消息丢失(如果没有设置ack为-1的话);以及其它异常。这时候,fill方法会调用KafkaUtils的getOffset方法,不过这个方法有些不符合useStartOffsetTimeIfOffsetOutOfRange的意思,即它并不是一定会从startOffsetTime配置中配置的offsetTime开始读。
public static long getOffset(SimpleConsumer consumer, String topic, int partition, KafkaConfig config) { long startOffsetTime = kafka.api.OffsetRequest.LatestTime(); if ( config.forceFromStart ) { startOffsetTime = config.startOffsetTime; } return getOffset(consumer, topic, partition, startOffsetTime); }
可以看出,如果没有设置forceFromStart,那么这个方法返回的offset将会是当前最大的offset,而忽略KafkaConfig中startOffsetTime的配置,使得PartitionManager直接跳到最新的消息开始处理。这样乍一看莫名其妙,但是试想,如果startOffsetTime对应的offset也out of range呢,这样KafkaSpout就陷入了死循环。而LatestOffsetTime()是始终存在的。但是,这样做而没有单独的配置,也没有日志记录说明这种权衡,会给用户带来麻烦。
在获取fetch到消息以后,获取的消息集可能会包含了各种例外情况,需要细致处理:
for (MessageAndOffset msg : msgs) { final Long cur_offset = msg.offset(); if (cur_offset < offset) { // Skip any old offsets. continue; } if (!had_failed || failed.contains(cur_offset)) { numMessages += 1; _pending.add(cur_offset);//_pending表示已经读取而未被ack的消息 _waitingToEmit.add(new MessageAndRealOffset(msg.message(), cur_offset)); _emittedToOffset = Math.max(msg.nextOffset(), _emittedToOffset); if (had_failed) {//说明此消息在failed集合里。从failed集合里remove掉它,因为它被重新加入了_waitingToEmit集合,将要被重新处理。 failed.remove(cur_offset);// } }
- 首先,需要考虑到FetchRequest指定的是返回集中最小的offset A,但是,实际上Kafka只保证返回的消息集中包括了offset为A的消息,这个消息集中也可能包括了offset比A更小的消息(由于压缩)。所以,fill方法首先要skip掉这些offset更小的消息
- 如果failed集合为空,fill方法就把得到的消息集中所有offset大于A的消息加入_waitingToEmit集合,同时加入_pending集合。然后把_emittedToOffset设为当前读取过的最大的offset。
- 如果读取到的消息在failed集合中,它在把这条消息加入_waitingToEmit集合与_pending集合后,还要把它从failed集合中去掉,否则这条消息就会永远在failed集合里。只有在fill方法中,failed集合中的元素才可能被移除,加入到_waitingToEmit集合,使它有机会被重新emit。
通过对fill方法的分析可以看到,如果一个消息始终fail,除非在PartitionManager的其它方法中把它移除,否则它会使PartitionManager的处理进度停止。下面将要看到,在fail和ack方法中,这样一直fail的消息还是有机会被丢弃的,但这取决于你的配置,而这些配置是很微妙的。
ack方法
ack方法的主要功能是把消息从_pending集合中去掉,表示这个消息处理完成。从_pending集合去除,PartitionManager才能获取正确的处理进度的信息,以更新Zookeeper里的记录。但是,它还有别的作用。
public void ack(Long offset) { if (!_pending.isEmpty() && _pending.first() < offset - _spoutConfig.maxOffsetBehind) { // Too many things pending! 已读取但未确认的消息太多了,就把此次确认的offset - maxOffsetBehind之前的清除了 _pending.headSet(offset - _spoutConfig.maxOffsetBehind).clear(); } _pending.remove(offset);//从_pending集合中移除它,表示这个消息已被处理 numberAcked++; }
当一个offset被ack时,ack方法会把所有小于offset - _spoutConfig.maxOffsetBehind的消息从_pending中移除。也就是说,即使这些被移除的消息失败了,也认为他们处理成功,使得在Zookeeper中记录的进度忽略这些被移除的消息。所以,假如task重启,那么这些失败但被移除出_pending集合的消息就不会被再处理。那么,这些失败了的消息,当Storm的acker发现它们处理失败,会发生什么呢?这由fail方法决定。
fail方法
public void fail(Long offset) { if (offset < _emittedToOffset - _spoutConfig.maxOffsetBehind) { LOG.info( "Skipping failed tuple at offset=" + offset + " because it‘s more than maxOffsetBehind=" + _spoutConfig.maxOffsetBehind + " behind _emittedToOffset=" + _emittedToOffset ); } else { LOG.debug("failing at offset=" + offset + " with _pending.size()=" + _pending.size() + " pending and _emittedToOffset=" + _emittedToOffset); failed.add(offset); numberFailed++; if (numberAcked == 0 && numberFailed > _spoutConfig.maxOffsetBehind) { throw new RuntimeException("Too many tuple failures"); } } }
当一个消息对应的tuple被fail时,fail方法会首先判断这个消息是否落后太多。如果它的offset小于(当前读取的最大offset - maxOffsetBehind), 那么就不把它加到failed集合里,使得它很可能不会被重新处理。如果不落后太多,就把它加到failed集合里,使得它可以被重新处理。如果没有消息ack,并且总的failed次数大于maxOffsetBehind,就抛出异常,代表PartitionManager工作出错,而这种情况只有在处理第一批消息并且这批消息的个数大于maxOffsetBehind时才可能发生。这样,有可能在某些情况下,使得PartitionManager卡住,但不会有异常。而且用numberFailed与spoutConfig.maxOffsetBehind比较,有些令人莫名其秒。
commit方法
commit方法被调用时,会调用lastCompletedOffset方法获取当前PartitionManager处理的进度,并且将这个进度持久化。这个“进度”,是说在此之前的所有消息都已被ack,或“不需要ack”, 总之,是说这些消息已处理完毕。
public long lastCompletedOffset() { if (_pending.isEmpty()) { return _emittedToOffset; } else { return _pending.first(); } }
在此,体现了_pending的作用。_pend中最小的元素,代表之前的元素都已处理完成。如果_pending为空,说明所有已被读取的元素都已处理完毕。
陷阱
failed方法,使得PartitonManager的有些行为非常隐晦。结合ack、fill和commit方法,可能会出现以下特殊情况,这些情况和KafkaConfig.maxOffBehind配置,及KafkaConfig.useStartOffsetTimeIfOffsetOutOfRange配置、KafkaConfig.fetchSizeBytes配置相关。
- maxOffsetBehind设置得较小,而fetchSizeBytes相对较大,使得maxOffsetBehind小于一次fetch得到的消息总数。设这批fetch得到的消息的offset范围为[a, b],那么所有小于(b - maxOffsetBehind)的offset的消息,即使处理失败,也不会被重试。设这样失败,但不会被重试的消息中的某个的offset为X, 那么如果某个大于( X + maxOffsetBehind)的消息被ack时,offset为X的这个消息会被从_pending集合中移除。但是如果所有大于(X + maxOffsetBehind)的消息都被fail了,而在(_emmittedToOffset与_emittedToOffset - maxOffsetBehind之间) 有消息failed了,那么failed集合中不会包括X,但会包括比X的offset大的元素,X不会被重试,但X会一直停留在_pending集合,造成commit无法更新实际进度,并且带来内存泄漏。
- 如果maxOffsetBehind比较大,就可能有failed的消息永远不会被忽略,而会一直重试,直到它成功。而fill方法使得在它成功之前,PartitionManager无法处理后续的消息。这样后续的blot必须保证正确地ack消息,否则这个partition就会卡在fail的这个消息上。
- 如果把useStartOffsetTimeIfOffsetOutOfRange设为true,同时forceFromStart设为false, 而startOffsetTime不为LatestTime,那么PartitonManager想要获取的消息out of range时,它会直接跳到LatestTime开始处理消息,而不会从startOffsetTime开始。这可能发生在Kafka进行了log retention之后。
- 如果有消息fail,那么这条消息之后的消息也可能被跟着重试,所以后边bolt的处理必须考虑到这种情况。
- 如果一条消息被拆成多个tuple发送,那么只要其中有一个tuple处理失败,这条消息产生的所有tuple就可能被重新发送。
总之,当前PartitionManager的实现还有很多需要改进之处,而且有些情况容易给用户带来困扰。
PS:关于KafkaSpout的进度,有个开源的工具可用:Capillary , 它用来对比当前KafkaSpout的处理进度和Kafka消息的LatestOffset。使用Play Framework和Scala开发。俺在此基础上做了些修改和补充,加入了Kafka集群的监控信息,特别是每个partition的leader,LatestOffset等信息,也加入图表用来显示未处理消息的累积情况,以及邮件报警。希望能有感兴趣的人一起把它的功能做得更完整一些。
转载于:
http://www.mamicode.com/info-detail-517847.html
相关推荐
《磁盘分区PartitionManager在Windows平台的应用与管理》 磁盘分区是计算机系统管理存储空间的重要方式之一,它能够帮助用户有效地组织和利用硬盘资源。PartitionManager是一款专为Windows平台设计的优秀磁盘分区...
《Paragon Partition Manager 11:硬盘分区管理的利器》 Paragon Partition Manager 11是一款功能强大的硬盘分区管理工具,由Paragon Software Group开发,主要用于帮助用户进行硬盘的分区操作,包括创建、删除、...
Paragon Partition Manager 是一套磁盘管理软件,是目前为止最好用的磁盘管理工具之一,能够优化磁盘使应用程序和系统速度变得更快,不损失磁盘数据下调整分区大小,对磁盘进行分区,并可以在不同的分区以及分区之间...
《EASEUS Partition Manager 9.1:硬盘分区管理的利器》 EASEUS Partition Manager 9.1是一款备受赞誉的硬盘管理工具,其最新版本为用户提供了强大的硬盘分区功能。这款软件以其高效、易用和全面的特性,赢得了广大...
《Paragon Partition Manager 10.0 汉化版:磁盘管理的高效工具》 Paragon Partition Manager 10.0 汉化版是一款广受好评的磁盘管理软件,它提供了丰富的功能,包括磁盘分区、合并、调整大小、迁移以及数据保护等。...
Paragon Partition Manager 10.0 是一款专业的磁盘分区管理软件,它的汉化版使得中文用户能够更方便地操作和理解软件的功能。这款工具在IT领域中扮演着重要的角色,因为它提供了对硬盘分区的全面控制,包括创建、...
《Paragon Partition Manager 10中文版:高效管理硬盘分区的得力助手》 Paragon Partition Manager 10中文版是一款强大的硬盘分区管理工具,专为Windows 7用户设计,提供无损的数据处理功能,帮助用户轻松进行硬盘...
Paragon Partition Manager是一款在IT行业中备受推崇的磁盘管理软件,专为个人用户和企业级用户提供高效、安全的硬盘管理解决方案。它集多种功能于一身,包括硬盘分区、分区格式转换、数据迁移、磁盘克隆等,是系统...
首先,Paragon Partition Manager 7.0 服务器版的核心功能之一是分区管理。它支持创建、删除、调整大小、移动和复制分区,使管理员能够在不影响系统运行的情况下灵活调整硬盘布局。这对于处理不断增长的数据量和适应...
Paragon Partition Manager 11是一款专业且功能强大的磁盘管理工具,它为用户提供了全面的磁盘维护、管理和优化解决方案。这款软件适用于个人用户以及中小型企业,能够帮助用户有效地管理和调整计算机硬盘分区,以...
### Paragon Partition Manager 7.0 使用教程详解 #### 一、引言 在计算机领域,硬盘分区是一项基本但至关重要的任务。对于那些寻求高效、安全地管理硬盘分区的用户而言,Paragon Partition Manager 7.0(以下简称...
Paragon Partition Manager 是一套磁盘管理软件,是目前为止最好用的磁盘管理工具之一,能够优化磁盘使应用程序和系统速度变得更快,不损失磁盘数据下调整分区大小,对磁盘进行分区,并可以在不同的分区以及分区之间...
**分区管理器9.0(Partition Manager 9.0)** Partition Manager 9.0是一款功能强大的磁盘分区管理工具,由Paragon Software Group开发,主要用于帮助用户在Windows操作系统下进行磁盘分区的操作和管理。它以其易用...
Paragon Partition Manager是一个类似于PQ PartitionMagic的磁盘分区工具集,是一套磁盘管理软件,PartitionMagic的最佳替代品!有着直觉的图形使用介面和支持鼠标操作。 【软件功能】 主要功能包括:能够不损失...
介绍:Paragon Partition Manager 是一套磁盘管理软件,是目前为止最好用的磁盘管理工具之一,能够优化磁盘使应用程序和系统速度变得更快,不损失磁盘数据下调整分区大小,对磁盘进行分区,并可以在不同的分区以及...
Paragon Partition Manager Server.rar
Paragon Partition Manager 7.0 服务器版是一款专业级的磁盘分区管理工具,专为服务器环境设计。这款软件提供了一系列强大的功能,旨在帮助系统管理员有效地管理和调整服务器上的磁盘空间,确保系统的稳定运行和数据...