数据源、storm应用、结果集,三者之间的关系
Storm是一个分布式的、可靠的、容错的数据流处理系统。使用storm应用场景可以稳定支撑着实时统计、对账、监控、数据同步、风控等多个业务。其中订单实时统计是需要实时跟进的,他对于数据准确性还有性能等各方面都有较高要求。我们可以通过对订单实时统计来描述使用storm时,所遇到的各个方面的问题,比如:可靠性、准确性以及它的性能等方面。
第一版:流程走通
在没有使用storm的时候,显示实时统计数据一般有两种方案:①在数据库里执行count、sum等聚合查询,这种是属于简单快速的实现方案,但缺点就是容易出现慢查询的情况;②在业务代码里对统计指标做累加,可以满足指标的快速查询,但统计逻辑耦合到业务代码,维护不方便,而且错误数据定位和修正不方便。所以如果想要解决指标快速查询或者解耦业务和统计,那基于storm的实时计算方案可以满足这两点需求。
一个storm应用的基本结构有三部分:数据源、storm应用、结果集。storm应用从数据源读取数据,经过计算后,把结果持久化或发送消息给其他应用。在数据源方面,最早尝试在业务代码里打日志的方式,但总有业务分支无法覆盖,采集的数据不全。我们的业务数据库是mysql,随后尝试基于mysql binlog的数据源,采用了阿里开源的canal,可以做到完整的收集业务数据变更。
在结果数据的处理上,我们把统计结果持久化到了mysql,并通过另一个后台应用的RESTful API对外提供服务,一个mysql就可以满足数据的读写需求。
但是为了提升实时统计应用吞吐量,需要提升消息的并发度。spout里设置了消息缓冲区,只要消息缓冲区不满,就会源源不断从消息源canal拉取数据,并把分发到多个bolt处理。
其中对于实时计算有个常见问题,就是通过订单实时统计的案例,可以抽象出一些基于storm实时计算的共性问题。
第二版:性能提升
为了确保数据的准确性,会把所有的统计指标持久化放在一个数据库事务里。一笔订单状态更新后,会在一个事务里有两类操作:
订单的历史状态也在数据库里存着,要与历史状态对比决定统计逻辑,并把最新的状态持久化。storm的应用本身是无状态的,需要使用存储设备记录状态信息。当大家知道实时计算好用后,各产品都希望有实时数据,统计逻辑越来越复杂。店铺、商品、用户等多个指标的写操作都是在一个事务里commit,这一简单粗暴的方式早期很好满足的统计需求,但是对于update操作持有锁时间过长,严重影响了并发能力。
为此做了数据库事务的瘦身(去除历史状态的mysql持久化,是通过单条binlog消息的前后状态对比,这样就做到了统计逻辑上的无状态。但又如何保证消息有且只有处理一次,为此引入了一个redis用于保存最近24小时内已成功处理的消息binlog偏移量,而storm的消息分发机制又可以保证相同消息总是能分配到一个bolt,避免线程安全问题。)
统计业务拆分,先是线上业务和公司内部业务分离,随后又把线上业务按不同产品拆分。这个不仅仅是bolt级别的拆分,而是在spout就完全分开。随着统计应用拆分,在canal和storm应用之间加上消息队列。canal不支持多消费者,而实时统计业务也不用关系数据库底层迁移、主从切换等维护工作,加上消息队列能把底层数据的维护和性能优化交给更专业的团队。
热点数据在mysql里做了分桶。比如,通常一个店铺天级别的统计指标在mysql里是一行数据。如果这个店铺有突发的大量订单,会出现多个bolt同时去update这行数据,出现数据热点,mysql里该行数据的锁竞争异常激烈。我们把这样的热点数据做了分桶,实验证明在特定场景下可以有一个数量级吞吐量提升。
所以第二版的主要变化在于引入了MQ,并使用redis作为消息状态的存储。而且由最初的一个应用,被拆成了多个应用。
第三版:准确性提升
经过第二版的优化,实时统计的吞吐量已经不成问题,但还是遇到了做大数据最重要的准确性的问题。
统计口径是会变化的,同样是GMV,一年前和现在的算法可能有变化。而实时统计只能按照当时的算法来做计算。有可能出现一段时间周期内的GMV,前一段是按旧算法来计算,后一段按新算法来计算,提供的数据就不准确了。所以实时统计难免会出现bug,有不准确的结果,修复错误数据是个难题。
为了解决这个问题,凡是涉及到两天以前数据的,一律由离线计算提供,最终展示给用户的数据,就是历史离线统计数据,并上今日昨日实时统计数据。因为离线统计有数据准备、建模、统计的过程,要花费几个小时,每天的凌晨很可能还得不到前一天的离线统计结果。一旦统计口径有变化,只需要重跑离线统计任务就可修复历史数据,做到了冷热数据分离。
消息状态管理:storm不提供消息状态管理,而且为了达到水平扩展,最好是消息之间无状态。对于大数据量、低精度的应用,需要做到无状态。而像订单实时统计这样数据量不算太大,但精度要求极高的场景,需要记录消息处理状态。而为了应付重启、分布式扩展的场景,往往需要额外的介质来存储状态。状态信息往往是kv形式的读写,我们在实际的应用中,使用过redis、HBase作为存储。
对于准确性要求高的场景,需要保证数据正确的只消费一次。storm的有三种消息处理模式:①若不实现ack和fail方法,无论后续处理结果如何,消息只会发送一次,必定不能满足高准确性;②若实现了ack和fail方法,只有调用了ack方法才会任务处理成功,否则会重试。可能会出现消息重复,在并发场景下重复又意味着可能出现乱序;③trident每个micro batch作为整体只成功处理一次,但也是无法保证消息真的只正确的处理一次,比如数据已经处理完毕并持久化,但向数据源ack时失败,就可能会有重试。
对于消息重复、乱序的场景,不是简单的消息幂等能解决,有以下的处理思路:①使用前面提到的状态管理的办法,识别出重复、乱序的数据;②业务逻辑中,兼容重复、乱序数据,比如维护一个业务状态机,把异常数据剔除。
对于时序判断,尽量不用使用时间戳,因为在分布式系统里,各服务器时间不一致是很常见的问题。
我们会尝试在运行过程中重启消息源、storm应用、存储/MQ等下游系统,或者制造网络丢包、延迟等异常,手工触发可能的消息丢失、重复、乱序场景,来验证我们的应用能否对应这些异常情况。
对于需要消息可靠处理的场景,是不适合复杂拓扑的,部分失败如何回滚,是否要全部bolt处理完毕才ack,是需要面对的问题。过长的拓扑链路,里面的慢速逻辑会拖慢整体性能。这时可以考虑使用更简化的拓扑,不同的逻辑之间尽量解耦,需要使用bolt的结果时,可以把数据持久化或者推送到MQ。
在生产环境中最必不可少的就是监控,除了服务器的基础监控,还加了不少storm特有的监控。
①消息延迟:消息在业务系统的时间戳与storm应用的当前时间戳对比,大于一定阈值则告警,不同应用的阈值会不同;
②消息处理时长、fail数:这两个都可以由storm的接口获取,数值偏大很可能是出了问题;
③应用TPS:记录应用的emit、ack、fail数的变化趋势,帮助分析应用的运行情况;
④任务级监控:每台服务器的worker、executor数量,这也可以通过storm接口获取。
除此之外,会有各类应用特有的监控,一般都是离线计算的结果与实时计算结果对比。对于数据同步类的应用,数据量比较大,可能会使用采样的方式做校验。