Flink 流式计算在节省资源方面的简单分析

阅读数:977 2019 年 11 月 12 日 10:34

Flink 流式计算在节省资源方面的简单分析

Flink 在小米的发展简介

Flink 流式计算在节省资源方面的简单分析

小米在流式计算方面经历了 Storm、Spark Streaming 和 Flink 的发展历程;从 2019 年 1 月接触 Flink 到现在,已经过去了大半年的时间了。对 Flink 的接触越深,越能感受到它在流式计算方面的强大能力;无论是实时性、时间语义还是对状态计算的支持等,都让很多之前需要复杂业务逻辑实现的功能转变成了简洁的 API 调用。还有不断完善的 Flink SQL 功能也让人充满期待,相信实时数据分析的门槛会越来越低,更多的业务能够挖掘出数据更实时更深入的价值。

在这期间,我们逐步完善了稳定性、作业管理、日志和监控收集展示等关系到用户易用性和运维能力的特性,帮助越来越多的业务接入到了 Flink。

流式作业管理服务的界面:

Flink 流式计算在节省资源方面的简单分析

Flink 作业的监控指标收集展示:

Flink 流式计算在节省资源方面的简单分析

Flink 作业异常日志的收集展示:

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 则没有类似问题):

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 的一些区别:

调度计算 VS 调度数据

对于任何一个分布式计算框架而言,如果“数据”和“计算”不在同一个节点上,那么它们中必须有一个需要移动到另一个所在的节点。如果把计算调度到数据所在的节点,那就是“调度计算”,反之则是“调度数据”;在这一点上 Spark Streaming 和 Flink 的实现是不同的。

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 节点信息:

复制代码
// KafkaRDD
override 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 流式计算在节省资源方面的简单分析

Flink 和 Storm 类似,都是通过“调度数据”来完成计算的,也就是“计算逻辑”初始化并启动后,如果没有异常会一直执行,源源不断地消费上游的数据,处理后发送到下游;有点像工厂里的流水线,货物在传送带上一直传递,每个工人专注完成自己的处理逻辑即可。

虽然“调度数据”和“调度计算”有各自的优势,但是在流式计算的实际生产场景中,“调度计算”很可能“有力使不出来”;比如一般流式计算都是消费消息队列 Kafka 或 Talos 的数据进行处理,而实际生产环境中为了保证消息队列的低延迟和易维护,一般不会和计算节点(比如 Yarn 服务的节点)混布,而是有各自的机器(不过很多时候是在同一机房);所以无论是 Spark 还是 Flink,都无法避免消息队列数据的跨网络传输。所以从实际使用体验上讲,Flink 的调度数据模式,显然更容易减少损耗,提高计算效率,同时在使用上更符合用户“直觉”,不易出现重复创建资源的情况。

不过这里不得不提的一点是,Spark Streaming 的“调度计算”模式,对于处理计算系统中的“慢节点”或“异常节点”有天然的优势。比如如果 Yarn 集群中有一台节点磁盘存在异常,导致计算不停地失败,Spark 可以通过 blacklist 机制停止调度计算到该节点,从而保证整个作业的稳定性。或者有一台计算节点的 CPU Load 偏高,导致处理比较慢,Spark 也可以通过 speculation 机制及时把同一计算调度到其他节点,避免慢节点拖慢整个作业;而以上特性在 Flink 中都是缺失的。

Mini batch vs streaming

Flink 流式计算在节省资源方面的简单分析

Spark Streaming 并不是真正意义上的流式计算,而是从批处理衍生出来的 mini batch 计算。如图所示,Spark 根据 RDD 依赖关系中的 shuffle dependency 进行作业的 Stage 划分,每个 Stage 根据 RDD 的 partition 信息切分成不同的分片;在实际执行的时候,只有当每个分片对应的计算结束之后,整个个 Stage 才算计算完成。

Flink 流式计算在节省资源方面的简单分析

这种模式容易出现“长尾效应”,比如如果某个分片数据量偏大,那么其他分片也必须等这个分片计算完成后,才能进行下一轮的计算 (Spark speculation 对这种情况也没有好的作用,因为这个是由于分片数据不均匀导致的),这样既增加了其他分片的数据处理延迟,也浪费了资源。

而 Flink 则是为真正的流式计算而设计的(并且把批处理抽象成有限流的数据计算),上游数据是持续发送到下游的,这样就避免了某个长尾分片导致其他分片计算“空闲”的情况,而是持续在处理数据,这在一定程度上提高了计算资源的利用率,降低了延迟。

Flink 流式计算在节省资源方面的简单分析

当然,这里又要说一下 mini batch 的优点了,那就在异常恢复的时候,可以以比较低的代价把缺失的分片数据恢复过来,这个主要归功于 RDD 的依赖关系抽象;如上图所示,如果黑色块表示的数据丢失(比如节点异常),Spark 仅需要通过重放“Good-Replay”表示的数据分片就可以把丢失的数据恢复,这个恢复效率是很高的。

Flink 流式计算在节省资源方面的简单分析

而 Flink 的话则需要停止整个“流水线”上的算子,并从 Checkpoint 恢复和重放数据;虽然 Flink 对这一点有一些优化,比如可以配置 failover strategy 为 region 来减少受影响的算子,不过相比于 Spark 只需要从上个 Stage 的数据恢复受影响的分片来讲,代价还是有点大。

总之,通过对比可以看出,Flink 的 streaming 模式对于低延迟处理数据比较友好,Spark 的 mini batch 模式则于异常恢复比较友好;如果在大部分情况下作业运行稳定的话,Flink 在资源利用率和数据处理效率上确实更占优势一些。

数据序列化

Flink 流式计算在节省资源方面的简单分析

简单来说,数据的序列化是指把一个 object 转化为 byte stream,反序列化则相反。序列化主要用于对象的持久化或者网络传输。

常见的序列化格式有 binary、json、xml、yaml 等;常见的序列化框架有 Java 原生序列化、Kryo、Thrift、Protobuf、Avro 等。

对于分布式计算来讲,数据的传输效率非常重要。好的序列化框架可以通过较低 的序列化时间和较低的内存占用大大提高计算效率和作业稳定性。在数据序列化上,Flink 和 Spark 采用了不同的方式;Spark 对于所有数据默认采用 Java 原生序列化方式,用户也可以配置使用 Kryo;而 Flink 则是自己实现了一套高效率的序列化方法。

首先说一下 Java 原生的序列化方式,这种方式的好处是比较简单通用,只要对象实现了 Serializable 接口即可;缺点就是效率比较低,而且如果用户没有指定 serialVersionUID 的话,很容易出现作业重新编译后,之前的数据无法反序列化出来的情况(这也是 Spark Streaming Checkpoint 的一个痛点,在业务使用中经常出现修改了代码之后,无法从 Checkpoint 恢复的问题);当然 Java 原生序列化还有一些其他弊端,这里不做深入讨论。

有意思的是,Flink 官方文档里对于不要使用 Java 原生序列化强调了三遍,甚至网上有传言 Oracle 要抛弃 Java 原生序列化:

Flink 流式计算在节省资源方面的简单分析

Flink 流式计算在节省资源方面的简单分析

相比于 Java 原生序列化方式,无论是在序列化效率还是序列化结果的内存占用上,Kryo 则更好一些(Spark 声称一般 Kryo 会比 Java 原生节省 10x 内存占用);Spark 文档中表示它们之所以没有把 Kryo 设置为默认序列化框架的唯一原因是因为 Kryo 需要用户自己注册需要序列化的类,并且建议用户通过配置开启 Kryo。

虽然如此,根据 Flink 的测试,Kryo 依然比 Flink 自己实现的序列化方式效率要低一些;如图所示是 Flink 序列化器(PojoSerializer、RowSerializer、TupleSerializer)和 Kryo 等其他序列化框架的对比,可以看出 Flink 序列化器还是比较占优势的:

Flink 流式计算在节省资源方面的简单分析

那么 Flink 到底是怎么做的呢?网上关于 Flink 序列化的文章已经很多了,这里我简单地说一下我的理解。

像 Kryo 这种序列化方式,在序列化数据的时候,除了数据中的“值”信息本身,还需要把一些数据的 meta 信息也写进去(比如对象的 Class 信息;如果是已经注册过的 Class,则写一个更节省内存的 ID)。

Flink 流式计算在节省资源方面的简单分析

但是在 Flink 场景中则完全不需要这样,因为在一个 Flink 作业 DAG 中,上游和下游之间传输的数据类型是固定且已知的,所以在序列化的时候只需要按照一定的排列规则把“值”信息写入即可(当然还有一些其他信息,比如是否为 null)。

Flink 流式计算在节省资源方面的简单分析

如图所示是一个内嵌 POJO 的 Tuple3 类型的序列化形式,可以看出这种序列化方式非常地“紧凑”,大大地节省了内存并提高了效率。另外,Flink 自己实现的序列化方式还有一些其他优势,比如直接操作二进制数据等。

凡事都有两面性,自己实现序列化方式也是有一些劣势,比如状态数据的格式兼容性(State Schema Evolution);如果你使用 Flink 自带的序列化框架序进行状态保存,那么修改状态数据的类信息后,可能在恢复状态时出现不兼容问题(目前 Flink 仅支持 POJO 和 Avro 的格式兼容升级)。

另外,用户为了保证数据能使用 Flink 自带的序列化器,有时候不得不自己再重写一个 POJO 类,把外部系统中数据的值再“映射”到这个 POJO 类中;而根据开发人员对 POJO 的理解不同,写出来的效果可能不一样,比如之前有个用户很肯定地说自己是按照 POJO 的规范来定义的类,我查看后发现原来他不小心多加了个 logger,这从侧面说明还是有一定的用户使用门槛的。

复制代码
// Not a POJO demo.public class Person { private Logger logger = LoggerFactory.getLogger(Person.class); public String name; public int age;}

针对这一情况我们做了一些优化尝试,由于在小米内部很多业务是通过 Thrfit 定义的数据,正常情况下 Thrift 类是通过 Kryo 的默认序列化器进行序列化和反序列化的,效率比较低。虽然官方提供了优化文档,可以通过如下方式进行优化,但是对业务来讲也是存在一定使用门槛;

复制代码
final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
// register the serializer included with Apache Thrift as the standard serializer
// TBaseSerializer states it should be initialized as a default Kryo serializer
env.getConfig().addDefaultKryoSerializer(MyCustomType.class, TBaseSerializer.class);

于是我们通过修改 Flink 中 Kryo 序列化器的相关逻辑,实现了对 Thrfit 类默认使用 Thrift 自己序列化器的优化,在大大提高了数据序列化效率的同时,也降低了业务的使用门槛。

总之,通过自己定制序列化器的方式,确实让 Flink 在数据处理效率上更有优势,这样作业就可以通过占用更低的带宽和更少的计算资源完成计算了。

本文小结

Flink 和 Spark Streaming 有非常大的差别,也有各自的优势,这里我只是简单介绍了一下自己浅薄的理解,不是很深入。不过从实际应用效果来看,Flink 确实通过高效的数据处理和资源利用,实现了成本上的优化;希望能有更多业务可以了解并试用 Flink,后续我们也会通过 Flink SQL 为更多业务提供简单易用的流式计算支持,谢谢!

参考文章:

1、《 Deep Dive on Apache Flink State 》 - Seth Wiesman

2、 Flink 原理与实现:内存管理

3、 Batch Processing — Apache Spark

评论

发布