1。背景ApacheFlink和ApacheStorm是业界广泛使用的两个分布式实时计算框架。其中，ApacheStorm（以下简称“Storm”）在美团点评的实时计算业务中应用较为成熟。它有管理平台，有通用的API和相应的文档，大量的实时操作是基于Storm构建的。ApacheFlink（以下简称“Flink”）近期备受关注。具有高吞吐、低延迟、高可靠、计算准确等特点，对事件窗口有很好的支持。有一定的应用。为了深入了解Flink框架，验证其稳定性和可靠性，评估其实时处理性能，找出系统中的不足，发现并优化其性能瓶颈，为用户提供最适合的实时计算engine，我们以实践经验丰富的Storm框架作为对照，进行了一系列的实验来测试Flink框架的性能，以及Flink作为实时计算框架的资源消耗，保证了“至少once”和“exactlyonce”的语义被计算出来。为Flink平台的选型、性能调优等决策和建设提出建议并提供数据支持，为后续SLA建设提供一定参考。Flink和Storm两种框架对比：2.测试目标评估Flink和Storm两种实时计算框架在不同场景和数据压力下的当前性能，获取它们的详细性能数据，找到处理性能的极限；了解不同配置对Flink性能影响的程度，分析各种配置的适用场景，进而提出调优建议。2.1测试场景“Input-Output”简单处理场景通过测试“Input-Output”等简单处理逻辑场景，尽量减少其他因素的干扰，体现两个框架的性能。同时衡量框架处理能力的极限，处理更复杂的逻辑性能不会比纯“输入输出”高。在用户的作业耗时较长的场景下，如果用户的处理逻辑比较复杂，或者访问数据库等外部组件，会增加执行时间，影响作业的性能。因此，我们测试了两个框架在用户作业耗时较长的场景下的调度性能。窗口统计场景实时计算往往需要对时间窗口或统计窗口进行统计，比如一天中每五分钟的访问次数，每百单使用了多少折扣等。Flink对窗口的支持比Storm更强大，API更完备，但我们也想知道两个框架在窗口统计的常见场景下的表现。对于精确计算场景（即消息传递语义为“exactlyonce”），Storm只能保证“AtMostOnce”和“AtLeastOnce”的消息传递语义，即可能存在重复发送。有很多业务场景对数据的准确性要求很高，希望消息传递不会重复或漏传。Flink支持“恰好一次”（ExactlyOnce）语义，但是在资源有限的情况下，更严格的精度要求可能会带来更高的成本，从而影响性能。因此，我们测试了两种框架在不同消息传递语义下的性能，希望能为精准计算场景下的资源规划提供数据参考。2.2性能指标吞吐量（Throughput）单位时间内计算框架成功传输的数据量。本次测试吞吐量的单位为：bar/second。它反映了系统的负载能力。在相应的资源条件下，系统单位时间内可以处理多少数据。吞吐量常用于资源规划，也用于辅助分析系统性能瓶颈，从而做出相应的资源调整，保证系统满足用户所需的处理能力。假设商家每小时可以做20份便当（吞吐量：20份/小时），一个外卖小哥每小时只能送两份便当（吞吐量：2份/小时），这个系统的瓶颈在外卖环节，可以安排十个送货员给商家送货。延迟数据从进入系统到流出系统所花费的时间。这个测试延迟的单位是毫秒。它反映了系统处理的实时性。金融交易分析等大量实时计算业务对时延要求很高。延迟越低，实时数据越强。假设商家做一份午餐需要5分钟，小哥送餐需要25分钟。在这个过程中，用户感受到了30分钟的延迟。如果更改配送计划后延迟变成60分钟，送来的时候食物是冷的，这个新计划是不能接受的。3、测试环境为本次测试搭建一个Standalone集群，分别为Storm和Flink搭建1个master节点和2个slave节点。其中，为了观察Flink在实际生产环境中的表现，部分测试内容也在Yarn环境中进行了测试。3.1集群参数3.2框架参数4.测试方法4.1测试流程数据生产DataGenerator以特定的速率产生数据，写入一个自增id和eventTime时间戳的KafkaTopic（TopicData）。数据处理StormTask和FlinkTask（每个测试用例不同）从同一个Offset开始消费KafkaTopicData，并将结果和对应的inTime和outTime时间戳分别写入两个Topic（StormTopic和FlinkTopic）。MetricsCollector根据outTime的时间窗口从这两个topic中收集测试metrics，每五分钟将对应的metrics写入MySQL表中。MetricsCollector根据outTime取一个五分钟的滚动时间窗口，计算五分钟平均吞吐量（输出数据条数），五分钟延迟的中位数（outTime-eventTime或outTime-inTime），99行等写入到MySQL对应的数据表中。最后计算MySQL表中吞吐量的平均值，选择延迟中位数和延迟99行的中位数，画图分析。4.2默认参数Storm和Flink默认都有AtLeastOnce语义。Storm开启ACK，ACKers个数为1。Flink的Checkpoint间隔为30秒，StateBackend默认为Memory。保证Kafka不是性能瓶颈，尽可能消除Kafka对测试结果的影响。在测试延迟时，数据生产率小于数据处理能力，假设数据写入Kafka后立即读取，即eventTime等于数据进入系统的时间。测试吞吐量时，从最老的KafkaTopic读取，假设这个Topic的测试数据量足够。4.3测试用例IdentityIdentity用例主要模拟简单的“输入-输出”处理场景，反映两个框架本身的性能。输入数据为“msgId,eventTime”，其中eventTime为数据产生时间。单次输入数据约20B，在进入job处理流程时记录inTime，在job处理完成后（准备输出时）记录outTime。Job从KafkaTopicData读取数据后，在字符串末尾附加时间戳，直接输出给Kafka。输出数据是“msgId,eventTime,inTime,outTime”。单次输出数据约50B。SleepSleep用例主要模拟用户作业耗时较长的场景，通过复杂的用户逻辑来体现框架差异的弱化，对比两种框架的调度性能。输入数据和输出数据都与身份相同。读取数据后，等待一定时间（1ms），然后在字符串末尾附加时间戳，输出WindowedWordCount。WindowedWordCount用例主要模拟窗口统计场景，反映两个框架在进行窗口统计时的性能差异。此外，它还用于测试精确计算场景，体现Flinkexactlyonedelivery的性能。输入为JSON格式，由msgId、eventTime和以空格分隔的一句话组成。单个输入数据约150B，读取数据后，解析JSON，然后将句子分成相应的词，发送给CountWindow，带有eventTime和inTime时间戳进行词数统计，记录一个窗口中最大和最小eventTime和inTime，最后用outTime时间戳Topic输出到Kafka。Spout/Source和OutputBolt/Output/Sink的并发度始终为1，增加并发度时只会增加JSONParser和CountWindow的并发度。由于Storm对windows的支持较弱，CountWindow是使用HashMap手动实现的，而Flink使用的是原生的CountWindow和对应的Reduce函数。5.测试结果5.1Identity单线程吞吐量上图中，蓝色柱为单线程Storm作业的吞吐量，橙色柱为单线程Flink作业的吞吐量。在Identity逻辑下，Storm的单线程吞吐量为8.7万条/秒，Flink的单线程吞吐量可达35万条/秒。当KafkaData的分区数为1时，Flink的吞吐量约为Storm的3.2倍；当KafkaData的分区数为8时，Flink的吞吐量约为Storm的4.6倍。由此可以看出，Flink的吞吐量大约是Storm的3-5倍。5.2Identity单线程job延迟使用outTime-eventTime作为延迟。图中蓝色虚线为Storm，橙色虚线为Flink。虚线是99线，实线是中位数。从图中可以看出，随着数据量的逐渐增加，Identity的延迟也逐渐增加。其中99线的增长速度快于中位数，Storm的增长速度快于Flink。其中QPS在8万以上的测试数据超过了Storm单线程的吞吐量，无法测试Storm，只有Flink的曲线。对比虚线最右端的数据，可以看出当StormQPS接近吞吐量时，延迟中值约为100毫秒，99线约为700毫秒，Flink的中值约为50毫秒，而99行大约是300毫秒。Flink在全吞吐量下的延迟大约是Storm的一半。5.3Sleep吞吐量从图中可以看出，当Sleep为1毫秒时，Storm和Flink的单线程吞吐量约为每秒900条，基本随着并发的增加呈线性增长。对比蓝色和橙色的柱子，可以发现此时两个框架的吞吐能力基本一致。5.4Sleep单线程作业延迟（中位数）仍然使用outTime-eventTime作为延迟。从图中可以看出，当Sleep为1毫秒时，Flink的延迟仍然低于Storm。5.5WindowedWordCount单线程吞吐量单线程执行一个大小为10的countwindow，吞吐量统计如图。从图中可以看出，Storm的吞吐量约为12000条记录/秒，FlinkStandalone的吞吐量约为43000条记录/秒。Flink的吞吐量仍然是Storm的3倍以上。5.6WindowedWordCountFlinkAtLeastOnce和ExactlyOnce吞吐量对比由于同一个算子的多个并行任务的处理速度可能不同，所以上游算子中不同快照中的内容在到达之前会被中间的并行算子处理下游运营商。subhours可以计算在同一个快照中。结果，这部分数据被重复处理。因此，Flink需要在ExactlyOnce语义下进行对齐，即在当前最早快照中的所有数据都处理完之前，不处理属于下一个快照的数据，而是在缓冲区中等待。目前测试用例中，JSONParser与CountWindow、CountWindow与Output之间需要对齐，需要一定的开销。为了体现对齐场景，Source/Output/Sink的并发量还是1，增加了JSONParser/CountWindow的并发量。具体流程详见前面的WindowedWordCount流程图。上图中橙色一栏是AtLeastOnce的吞吐量，黄色一栏是ExactlyOnce的吞吐量。两者对比可以看出，在当前并发条件下，ExactlyOnce的吞吐量比AtLeastOnce低6.3%，每条消息发送时自动ACK，不再等待bolt的ACK，也不重新发送消息，这是AtMostOnce语义。上图中蓝色柱是AtLeastOnce的吞吐量，浅蓝色柱是AtMostOnce的吞吐量。两者对比可以看出，在当前并发条件下，AtMostOnce语义下的吞吐量比AtLeastOnce高16.8%。作业处理时间短或Thread.sleep()精度不高，outTime-inTime为零或无比较意义；WindowedWordCount可以有效的衡量outTime-inTime的值，并将其绘制在与outTime-eventTime同一张图上，其中outTime-eventTime为虚线，outTime-InTime为实线。观察两条橙色虚线，我们可以发现Flink两种方式统计的延迟都维持在较低的水平；观察两条蓝色曲线，我们可以发现Storm的outTime-inTime较低，而outTime-eventTime一直很高，也就是说inTime和eventTime的差值一直很大，这可能与Storm和eventTime的方式有关Flink读取数据。蓝色虚线表示Storm的延迟随着数据量的增加而增加，而橙色虚线表示Flink的延迟随着数据量的增加而减少（这里不测Flink的吞吐量，延迟Flink仍然接近吞吐量。会上升）。即使你只关注outTime-inTime（即图中的实线），你仍然可以发现，当QPS逐渐增加时，Flink在延迟方面的优势开始体现出来。5.9WindowedWordCountFlinkAtLeastOnce和ExactlyOnce延迟比较图中黄线为99行，橙线为中位数，虚线为AtLeastOnce，实线为ExactlyOnce。图中对应颜色的虚线和实线基本重合。可以看出，FlinkExactlyOnce的延迟中值曲线与AtLeastOnce基本吻合，延迟性能没有太大差异。5.10WindowedWordCountStormAtLeastOnce和AtmostOnce延迟对比图蓝线为99行，淡蓝色线为中位数，虚线为AtLeastOnce，实线为AtMostOnce。当QPS为4000及之前时，虚线和实线基本重合；QPS为6000时，两者有区别，虚线略高；当QPS接近8000时，已经超过了AtLeastOnce语义下Storm的吞吐量，所以只有实线点。可以看出，当QPS低的时候，StormAtMostOnce和AtLeastOnce在延迟上没有区别。随着QPS的增加，差异开始变大，AtMostOnce的延迟较低。5.11Flink不同StateBackends的WindowedWordCount吞吐量对比Flink支持Standalone和onYarn集群部署模式，支持Memory、FileSystem和RocksDB三种状态存储后端（StateBackends）。由于线上运行的需要，测试了三种StateBackend在两种集群部署模式下的性能差异。其中，Standalone的存储路径是JobManager上的一个文件目录，onYarn的存储路径是HDFS上的一个文件目录。对比三组列，可以发现FileSystem和Memory的吞吐量相差不大，RocksDB的吞吐量只有其他两者的十分之一左右。对比两种颜色可以发现，Standalone和onYarn的整体差别并不大。使用FileSystem和Memory时，onYarn模式的吞吐量略高，使用RocksDB时，Standalone模式的吞吐量略高。5.12WindowedWordCountFlink中不同StateBackends的延迟比较当使用FileSystem和Memory作为Backends时，延迟基本相同并且更低。使用RocksDB作为Backends时，延迟略高，由于吞吐量较低，达到吞吐量瓶颈前的延迟会急剧增加。其中，在onYarn模式下吞吐量较低，接近吞吐量时延迟较高。6.结论与建议6.1框架性能从5.1和5.5的测试结果可以看出，Storm的单线程吞吐量约为87000条/秒，Flink的单线程吞吐量可达350000条/秒第二。Flink的吞吐量大约是Storm的3-5倍。从5.2和5.8的测试结果可以看出，在吞吐量接近的情况下，StormQPS的中位延迟（包括Kafka读写时间）在100毫秒左右，99行在700毫秒左右。Flink的中位数约为50毫秒，99行约为300毫秒。Flink在全吞吐时的延迟大约是Storm的一半，随着QPS的逐渐提升，Flink在延迟上的优势开始显现。综上所述，Flink框架本身的性能优于Storm。6.2对比5.1和5.3、5.2和5.4的测试结果可以发现，当单个bolt的休眠时长达到1毫秒时，Flink的延迟仍然低于Storm，但是吞吐量优势基本没有反映。因此，用户逻辑越复杂，耗时越长，测试该逻辑所体现的框架差异越小。6.3不同消息传递语义的差异从5.6、5.7、5.9、5.10的测试结果可以看出，FlinkExactlyOnce的吞吐量比AtLeastOnce低6.3%，延迟差异不大;StormAtMostOnce语义下的吞吐量与AtLeastOnce相比，高出16.8%，延迟略低。由于Storm会对每条消息进行ACK，因此Flink是基于一批消息的检查点。不同的实现原理导致两者在AtLeastOnce语义上的开销差异较大，影响性能。但是Flink对ExactlyOnce语义的实现只是增加了对齐操作，所以在算子的并发量不大，没有出现慢节点的情况下，对Flink的性能影响不大。StormAtMostOnce语义下的性能仍然低于Flink。6.4Flink状态存储后端选择Flink提供了内存、文件系统、RocksDB三种StateBackend，结合5.11和5.12的测试结果，三者的对比如下：6.5推荐使用Flink的场景基于以上测试结果，以下实时计算场景推荐考虑使用Flink框架进行计算：要求消息传递语义为ExactlyOnce的场景；数据量大、高吞吐、低延迟的场景；需要状态管理或窗口统计的场景。七、展望本次测试还有部分内容没有深入测试，需要后续测试补充。比如：ExactlyOnce的吞吐量会不会随着并发的增加而明显下降？当用户耗时达到1ms时，框架的差异不再明显（Thread.sleep()的精度只能做到毫秒级），那么在什么范围内Flink的优势还能体现出来呢？本次测试只观察了吞吐量和时延两个指标，并没有关注统计数据层面的系统可靠性、可扩展性等重要性能指标，需要后期补充。使用RocksDBStateBackend的Flink吞吐量较低，需要进一步探索和优化。Flink更高级的API，比如TableAPI&SQL、CEP等，还需要进一步理解和完善。【本文为栏目机构“美团点评技术团队”原创稿件，转载请微信联系机构♂获得授权】点此查看作者更多好文
