全链路监控扩展篇:storm从1.0.2 升级到 1.2.2全流程-下

导 读

承接上文,本文主要介绍业务代码升级相关内容。业务代码升级,主要是从旧的基于Zookeeper的kakfa消费方式,切到新的消费者API上。

代码调整

1

调整1

使用新版storm-kafka-clien包代替老版本使用的storm-kafka(此处需要特别注意,网上很多demo都是基于老版本的storm写的,所以引用的包还是storm-kafka)。

即使用

代替

2

调整2

storm-kafka包升级以后,消费kafka的配置方式也发生了变化。

1.0.2版本的配置方式

当前的版本storm会将offset写入到zookeeper中,根据offset从broker中获取数据。每个Spout中包含一个consumer消费者,通过SpoutConfig完成相应的配置。代码如下:

相关参数见下表:

1.2.2版本的配置方式

当前版本中会将 kafka消息的 topic, partition, offset, key, value作为一个 ConsumerRecord 发送,当我们仅需要 value时,需要通过 SimpleRecordTranslator 对数据进行过滤。

当前版本中摒弃了之前版本中通过startOffsetTime, scheme, fetchSizeBytes, bufferSizeBytes, fetchMaxWait, clientId 等配置字段,通过setProp的方式作为kafka的配置参数,配置方式如下图所示。

相关参数见下表:

要点总结与归纳

1

使用新版的storm-kafka-client

storm-kafka 是老版本的包,官方推荐使用storm-kafka-client 代替。凡是网上搜到的代码中,使用storm-kafka ,还是使用基于ZK模式消费kafka,并没有使用 kafka新消费者API。两者在使用方式上不相同,需要注意。

2

切换时如何保证数据不丢失?

在升级过程中,我们希望能够做到平滑升级,保证数据不丢失。

方案一

如果在后半夜或者周末的时候业务会暂停,没有新的kafka消息进入,数据切换就很简单。只需要新版的Consumer从最近的offset开始消费就可以。

推荐将 setFirstPollOffsetStrategy配置为 UNCOMMITTED_LATEST,当业务启动数据进入后,即开始消费。

方案二

如果业务数据是不间断的,又不希望数据丢失,需要将将zk中存储的offset,写入到为“_consumer_offsets”的topic中(在0.10.x以后的kafka版本中,将offset写入到了_consumer_offsets的topic中),storm官方提供了相应的升级方案。

升级小花絮

在整个升级过程中,我们还遇到了一个问题,即升级后,新版本storm使用kafka新消费者无法获取数据,但是切回老版本ZK模式,却没有问题。

经过相关同事排查,在kafka服务器上,我们看到有报错日志如下:

The replication factor for the offsets topic (set higher to ensure availability). Internal topic creation will fail until the cluster size meets this replication factor requirement.

即这个参数的意思是内部topic副本配置数量,如果集群数量(broker)小于副本数量,topic创建会失败。

内部topic是指什么? 老版本kafka消费者,是基于ZK的,offset数据也保存在ZK上面;新版本kafka,offset都保存在broker上面的topic,即内部topic。

早期kafka搭建时,使用版本是0.8.1,只搭建了2个broker,虽然后面升级到0.11.0,但是由于一直使用基于ZK的消费模式,导致问题一直没用报出来,直到这次使用新消费者,才发现问题。

问题清楚后,我们扩展了kafka集群节点数量,从2个扩展到3个,随后新版storm 可以正常消费数据。

至此storm从1.0.2 升级到 1.2.2全流程完结。

  • 发表于:
  • 原文链接https://kuaibao.qq.com/s/20180927G14KM400?refer=cp_1026
  • 腾讯「云+社区」是腾讯内容开放平台帐号(企鹅号)传播渠道之一,根据《腾讯内容开放平台服务协议》转载发布内容。
  • 如有侵权,请联系 yunjia_community@tencent.com 删除。

扫码关注云+社区

领取腾讯云代金券