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