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

Flink在小米的发展简介

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

在这期间，我们逐步完善了稳定性、作业管理、日志和监控收集展示等关系到用户易用性和运维能力的特性，帮助越来越多的业务接入到了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则没有类似问题）:

• 对于简单的从消息队列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的实现是不同的。

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

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

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

Mini batch vs streaming

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

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

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

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

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

总之，通过对比可以看出，Flink的streaming模式对于低延迟处理数据比较友好，Spark的mini batch模式则于异常恢复比较友好；如果在大部分情况下作业运行稳定的话，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原生序列化：

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

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

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

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

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

如图所示是一个内嵌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