![]()
关于Flink流式计算节省资源方面你必须知道的技巧 小米在流式计算方面经历了Storm、Spark Streaming和Flink的发展历程；从2019年1月接触Flink到现在，已经过去了大半年的时间了。对Flink的接触越深，越能感受到它在流式计算方面的强大能力；无论是实时性、时间语义还是对状态计算的支持等，都让很多之前需要复杂业务逻辑实现的功能转变成了简洁的API调用。还有不断完善的Flink SQL功能也让人充满期待，相信实时数据分析的门槛会越来越低，更多的业务能够挖掘出数据更实时更深入的价值。在这期间，我们逐步完善了稳定性、作业管理、日志和监控收集展示等关系到用户易用性和运维能力的特性，帮助越来越多的业务接入到了Flink。Spark Streaming 迁移到Flink的效果小结在业务从Spark Streaming迁移到Flink的过程中，我们也一直在关注着一些指标的变化，比如数据处理的延迟、资源使用的变化、作业的稳定性等。其中有一些指标的变化是在预期之中的，比如数据处理延迟大大降低了，一些状态相关计算的“准确率”提升了；但是有一项指标的变化是超出我们预期的，那就是节省的资源。信息流推荐业务是小米从Spark Streaming迁移到Flink流式计算最早也是使用Flink最深的业务之一，在经过一段时间的合作优化后，对方同学给我们提供了一些使用效果小结，其中有几个关键点：
- 对于无状态作业，数据处理的延迟由之前Spark Streaming的16129ms降低到Flink的926ms，有94.2%的显著提升（有状态作业也有提升，但是和具体业务逻辑有关，不做介绍）；
- 对后端存储系统的写入延迟从80ms降低到了20ms左右，如下图（这是因为Spark Streaming的mini batch模式会在batch最后有批量写存储系统的操作，从而造成写请求尖峰，Flink则没有类似问题）:
- 对于简单的从消息队列Talos到存储系统HDFS的数据清洗作业（ETL），由之前Spark Streaming的占用210个CPU Core降到了Flink的32个CPU Core，资源利用率提高了84.8%；
其中前两点优化效果是比较容易理解的，主要是第三点我们觉得有点超出预期。为了验证这一点，信息流推荐的同学帮助我们做了一些测试，尝试把之前的Spark Streaming作业由210个CPU Core降低到64个，但是测试结果是作业出现了数据拥堵。这个Spark Streaming测试作业的batch interval 是10s，大部分batch能够在8s左右运行完，偶尔抖动的话会有十几秒，但是当晚高峰流量上涨之后，这个Spark Streaming作业就会开始拥堵了，而Flink使用32个CPU Core却没有遇到拥堵问题。
很显然，更低的资源占用帮助业务更好的节省了成本，节省出来的计算资源则可以让更多其他的业务使用；为了让节省成本能够得到“理论”上的支撑，我们尝试从几个方面研究并对比了Spark Streaming和Flink的一些区别：对于任何一个分布式计算框架而言，如果“数据”和“计算”不在同一个节点上，那么它们中必须有一个需要移动到另一个所在的节点。如果把计算调度到数据所在的节点，那就是“调度计算”，反之则是“调度数据”；在这一点上Spark Streaming和Flink的实现是不同的。Spark的核心数据结构RDD包含了几个关键信息，包括数据的分片（partitions)、依赖（dependencies)等，其中还有一个用于优化执行的信息，就是分片的“preferred locations”// RDD/** * Optionally overridden by subclasses to specify placement preferences. */protected def getPreferredLocations(split: Partition): Seq[String] = Nil
这个信息提供了该分片数据的位置信息，即所在的节点；Spark在调度该分片的计算的时候，会尽量把该分片的计算调度到数据所在的节点，从而提高计算效率。比如对于KafkaRDD，该方法返回的就是topic partition的leader节点信息：// KafkaRDDoverride def getPreferredLocations(thePart: Partition): Seq[String] = { val part = thePart.asInstanceOf[KafkaRDDPartition] Seq(part.host) // host: preferred kafka host, i.e. the leader at the time the rdd was created }
”调度计算”的方法在批处理中有很大的优势，因为“计算”相比于“数据”来讲一般信息量比较小，如果“计算”可以在“数据”所在的节点执行的话，会省去大量网络传输，节省带宽的同时提高了计算效率。但是在流式计算中，以Spark Streaming的调度方法为例，由于需要频繁的调度”计算“，则会有一些效率上的损耗。首先，每次”计算“的调度都是要消耗一些时间的，比如“计算”信息的序列化 → 传输 → 反序列化 → 初始化相关资源 → 计算执行→执行完的清理和结果上报等，这些都是一些“损耗”。另外，用户的计算中一般会有一些资源的初始化逻辑，比如初始化外部系统的客户端（类似于Kafka Producer或Consumer)；每次计算的重复调度容易导致这些资源的重复初始化，需要用户对执行逻辑有一定的理解，才能合理地初始化资源，避免资源的重复创建；这就提高了使用门槛，容易埋下隐患；通过业务支持发现，在实际生产过程中，经常会遇到大并发的Spark Streaming作业给Kafka或HBase等存储系统带来巨大连接压力的情况，就是因为用户在计算逻辑中一直重复创建连接。Spark在官方文档提供了一些避免重复创建网络连接的示例代码，其核心思想就是通过连接池来复用连接：rdd.foreachPartition { partitionOfRecords =>// ConnectionPool is a static, lazily initialized pool of connections val connection = ConnectionPool.getConnection() partitionOfRecords.foreach(record => connection.send(record)) ConnectionPool.returnConnection(connection) // return to the pool for future reuse }}
需要指出的是，即使用户代码层面合理的使用了连接池，由于同一个“计算”逻辑不一定调度到同一个计算节点，还是可能会出现在不同计算节点上重新创建连接的情况。Flink和Storm类似，都是通过“调度数据”来完成计算的，也就是“计算逻辑”初始化并启动后，如果没有异常会一直执行，源源不断地消费上游的数据，处理后发送到下游；有点像工厂里的流水线，货物在传送带上一直传递，每个工人专注完成自己的处理逻辑即可。虽然“调度数据”和“调度计算”有各自的优势，但是在流式计算的实际生产场景中，“调度计算”很可能“有力使不出来”；比如一般流式计算都是消费消息队列Kafka或Talos的数据进行处理，而实际生产环境中为了保证消息队列的低延迟和易维护，一般不会和计算节点（比如Yarn服务的节点）混布，而是有各自的机器（不过很多时候是在同一机房）；所以无论是Spark还是Flink，都无法避免消息队列数据的跨网络传输。所以从实际使用体验上讲，Flink的调度数据模式，显然更容易减少损耗，提高计算效率，同时在使用上更符合用户“直觉”，不易出现重复创建资源的情况。不过这里不得不提的一点是，Spark Streaming的“调度计算”模式，对于处理计算系统中的“慢节点”或“异常节点”有天然的优势。比如如果Yarn集群中有一台节点磁盘存在异常，导致计算不停地失败，Spark可以通过blacklist机制停止调度计算到该节点，从而保证整个作业的稳定性。或者有一台计算节点的CPU Load偏高，导致处理比较慢，Spark也可以通过speculation机制及时把同一计算调度到其他节点，避免慢节点拖慢整个作业；而以上特性在Flink中都是缺失的。