Flink引擎深度优化与生产实践摘要：本文整理自快手实时计算团队技术专家刘建刚在FlinkForwardAsia的演讲。2021生产实践课。主要内容包括：快手Flink的历史与现状Flink容错改进Flink引擎控制与实践快手批处理实践未来规划01快手Flink的历史与现状快手Flink自2018年以来的深度整合After4经过多年的发展，实时计算平台逐渐完善和赋能各种外围组件。2018年，我们对Flink1.4进行了平台化建设，大大提升了运维管理能力，使其可用于生产。2019年我们在1.6版本的基础上开始迭代开发，很多业务开始实时化。例如，优化IntervalJoin为商业化等平台带来显着收益，开发实时多维分析加速超多维报表的实时化。今年，我们的FlinkSQL平台也在用。2020年，我们升级到1.10，完善了SQL的多项功能，进一步优化了Flink的核心引擎，保证了Flink的易用性、稳定性和可维护性。2021年开始聚焦离线计算，支持一体化湖仓建设，进一步完善Flink生态。上图是快手基于Flink的技术栈。核心底层是Flink的计算引擎，包括流计算和批处理。我们在稳定性和性能方面做了很多工作。外层是与Flink打交道的外围组件，包括Kafka、rocketMQ等中间件，ClickHouse、Hive等数据分析工具，以及Hudi等数据湖的使用。用户可以基于Flink和这些组件构建各种应用，涵盖实时、准实时、批处理的各种场景。最外层是具体的使用场景。电商、商业化等常见视频相关业务方。应用场景包括机器学习、多维分析等。另外还有很多技术部门基于Flink实现数据的导入和转换，比如CDC、湖仓集成等。在应用规模上，我们有50万个CPU核，我们主要通过Yarn和K8s来管理资源。上面运行着2000+个作业，峰值处理速度6亿/秒，日处理量31.7万亿。活动期间流量甚至会翻倍。02容错提升容错主要包括以下几个部分：第一，单点恢复，支持任意数量任务失败时就地重启，长时间运行的作业基本可以连续；二是集群故障响应，包括冷备、热备、Kafka双集群集成；最后，使用黑名单。为了实现exactly-once，Flink需要在任何节点发生故障时重启整个作业。全局重启会带来很长的停顿，最多十分钟。有些场景不追求exactly-once，比如推荐等实时场景，但是对服务可用性要求高，不能容忍作业中断。还有模型训练等初始化比较慢的场景，重启时间特别长。会产生很大的影响。基于以上考虑，我们开发了单点恢复功能。上图是单点恢复的基本原理。如图，一共有三个任务，中间任务失败。首先，Flink的master节点会重新调度中间任务。此时上下游任务不会失败，而是等待重连。中间任务调度成功后，主节点会通知下游任务重新连接上游任务。同时，中间任务也会连接到它的上游任务，通过重建读视图来恢复数据读取。上下游连接成功后，作业就可以正常工作了。了解了基本原理之后，我们来看一下在线多任务恢复的案例。在实际环境中，经常会出现多个任务同时失败的情况。这个时候我们就把失败的任务按照拓扑顺序一个一个的恢复。比如上图中，recovery是从左到右的。该功能上线后，我们内部已有近百家运营商使用了该功能。在正常故障下，可以连续流式运行。即使是很小的流量波动，业务也可以毫无察觉，业务方彻底告别服务中断。流式噩梦。一旦发生集群故障，那是致命的，所有数据都会丢失，服务也会挂掉。我们的解决方案主要包括冷备、热备以及Flink和Kafka的双集群集成。冷备份主要是指对数据进行备份，以便集群宕机后可以在另一个集群上快速启动计算任务。如上图，KwaiJobManager是快手的作业管理服务，failovercoordinator主要负责故障处理。我们会将所有jar包和其他文件保存在HDFS中，所有信息保存在Mysql中，两者都是高可用的。作业运行在主集群ClusterA上，线上使用增量快照，会造成文件依赖问题，所以我们定期做save??points，复制到备集群。为了避免文件过多，我们设置了定时删除历史快照。一旦服务检测到集群A出现故障，会立即启动集群B中的作业，并从最新的快照恢复，保证状态不丢失。对于用户来说，只需要搭建好主备集群，剩下的交给平台，全程故障用户感知不到。双机热备是指两个集群同时运行同一个任务。我们的热备是全链路的，Kafka或者ClickHouse是双跑的。顶层显示层只会使用其中一个结果数据进行显示。一旦发生故障，显示层会立即切换到另一个数据。切换过程不到一秒，全程用户无感觉。与冷备份相比，热备份需要相同数量的资源进行备份操作，但切换速度更快，更适合春晚等要求极高的场景。Flink与Kafka的双集群融合主要是因为快手的Kafka具备双集群的能力，所以Flink需要支持读写双集群的Kafka主题，这样Flink在一个Kafka时可以无缝切换上线群集已关闭。如上图所示，我们的Flink抽象出了Kafka双集群。一个逻辑主题底层对应两个物理主题，由多个分区组成。Flink消费逻辑topic相当于同时读取最底层的两个底层。物理主题的数据。对于集群的各种变化，我们都抽象为分区上的扩容和缩容。比如集群挂了，可以认为是逻辑topic的partitionreduction；如果将一个单独的簇分割成两个簇，则可以看作是逻辑题目的扩展；主题迁移可以看作是逻辑主题扩展然后收缩。这里我们以双集群为例。其实无论是双集群还是多集群，原理都是一样的，我们都提供支持。当出现以下两种情况时，需要使用黑名单功能。一是反复调度故障机，导致作业频繁失败。另一种是由于硬件或网络等原因，导致部分Flink节点卡死但未失效。对于第一种情况，我们开发了阈值停电。如果job在同一台机器上失败或者多次部署阈值失败，超过配置的阈值会拉黑；对于第二种情况，我们建立了异常分类机制。对于网络卡顿、磁盘卡顿，直接驱逐容器、堵机。此外，我们还对外开放了阻塞接口，打通了O&MYarn等外部系统，实现了实时阻塞。我们也以Flink黑名单为契机，建立了一套完整的硬件异常处理流程，实现了作业自动迁移，全程自动运维，用户无感知。03Flink引擎控制与实践3.1Flink实时控制针对长时间运行的实时作业，用户经常需要进行调整参数等变更行为，以及作业降级、修改日志级别等一些系统运维等等，这些修改需要重启作业需要几分钟到几十分钟才能生效。在一些重要场合，这是不能容忍的。例如，在某项活动中，或在故障排除的关键点，一旦停工，工作就会功亏一篑。因此，我们需要在不停止运行的情况下实时调整运行的行为，即实时控制。从更广泛的角度来看，Flink不仅仅是一个计算任务，更是一个长期运行的服务。我们的实时控制就是基于这样的考虑，为实时计算提供一种交互式的控制方式。如上图所示，用户通过经典的kv数据类型与FlinkDispatcher进行交互。Flink收到消息后，首先会持久化到zk中进行failover，然后根据具体的消息进行相应的控制，比如控制resourcemanager、Controljobmaster或者其他组件。我们不仅支持用户自定义动态参数，还为用户提供了许多现成的系统控件。用户自定义主要是使用RichFunction获取动态参数，并实现相应的逻辑，从而在作业运行时实时传入参数，达到实时控制的效果。系统提供的实时管控能力主要包括数据源限速、采样、重置Kafka偏移量、调整快照参数、更改运维相关日志级别、拉黑节点等功能。此外，我们还支持动态修改部分Flink原生配置。快手内部实时控制功能已经商业化，非常方便用户使用。3.2Source端管控能力当Flink处理历史任务或作业性能跟不上时，会造成以下问题：第一，Source的并发处理速度不一致，会进一步加剧数据乱序、丢失、丢失等问题。慢对齐。其次，快照的大小会不断增长，严重影响作业性能。另外还有不可控的流量资源，在高负载情况下会出现CPU吃饱、OOM等稳定性问题。由于Flink是流水线实时计算，从数据源入手可以从根本上解决问题。首先我们来看看历史数据的精准回放功能。上图是将Kafka的历史数据以一倍的速率进行消费，Flink作业赶上滞后可以转为实时消费。这样就可以有效解决复杂任务的稳定性问题。上图中的公式是一个基本原理。消费倍数=Kafka时间差/Flink系统时间差。用户在使用时只需配置乘法器即可。另一个能力是QPS限速。当数据流量较大时，会导致Flink负载过高，运行不稳定。基于令牌桶算法，我们实现了一套分布式限速策略，可以有效降低Flink的压力。使用QPSratelimit后，作业变得非常健康，在上图中以绿色可见。对于2019年的春晚大屏，我们通过这项技术实现了灵活性和易用性的保证。此外，我们还支持分区变化的自动适配和实时控制，让用户随时随地调整作业的QPS。最后一个功能是数据源对齐，主要是指水印的对齐。首先，每个子任务会定期向主节点报告自己的水印进度，主要包括水印的大小和速度。主节点会计算出下一个周期的目标，也就是期望的最大watermark，然后返回一个diff给各个节点。每个源任务都会保证下一个周期的watermark不超过设定的目标。上图最下面是target的计算公式，预测每个task下一个循环结束时的waterMark值，加上我们允许的maxdiff然后取最大值。这样可以保证各个源的进度一致，避免diff过大。大的稳定性问题。3.3BalancedJobScheduling资源不平衡经常发生在生产环境中。比如第一点，Flink的任务分布不均，导致任务管理器资源使用不均，作业的性能往往受限于最繁忙的节点。针对这个问题，我们制定了作业平衡调度策略；第二点是CPU使用率不均衡，有的机器满有的机器闲。为了解决这个问题，我们开发了CPU均衡调度的功能。上图中有3个jobVertex，通过hashshuffle连接起来。上图中间部分展示了Flink的调度。每个jobVertex从上到下将任务调度到slot。结果，前两个插槽已满，其他插槽空闲。第一个任务管理器已满，第二个已满。任务管理器空闲。这是一个典型的资源倾斜场景，我们对其进行了优化。调度时，先计算需要的资源总量，即需要多少个taskmanager，然后计算每个TM分配的slot数量，保证TM中的slot资源均衡。最后将任务平均分配到各个slot中，保证slot中的任务是均衡的。实际运行过程中还有一种倾斜——CPU倾斜，我们来看看如何解决这个问题。上图左侧，用户申请一核实际只用了0.5核，也有申请一核实际用了1核的。按照默认的调度策略，这种情况大量出现可能会导致部分机器CPU占用率高，而其他机器空闲。负载高的机器性能和稳定性会很差。那么如何让application和use之间的差异尽可能小呢？我们的解决方案是准确描述工作资源。具体方法分为以下几步：在job的运行过程中，统计每个task所在容器的CPU使用率，然后创建task到executionSlotSharingGroup，再到container的映射，这样我们知道每个任务所在slot的CPU使用率，然后根据映射关系重启job，根据task所在slot的历史CPU使用率申请相应的资源。一般来说，一些缓冲区是保留的。如上右图所示，如果预测足够准确，重启后任务管理器使用的资源不变，但应用价值变小，两者差距变小。其实业界一些比较先进的系统，比如borg，是支持应用值动态修改的，但是我们底层调度资源不支持这种策略，所以只能在Flink层使用资源画像来解决这个问题。当然，资源画像不能保证100%准确。我们还有其他策略，例如限制资源持续分配给CPU负载高的机器，以最大限度地减少不平衡。此外，我们还建立了分级保障体系。不同优先级的作业有不同的cgroup限制。例如，低优先级的作业不再超配，高优先级的作业允许稍微超配，避免CPU使用率过高导致不平衡。04快手批处理实践上图是我们的批处理架构图。最底层是离线集群，中间是Flink引擎和Flink的数据流API，SQLAPI，上面是一些平台方，比如sql入口，时序调度平台等。另外还有一些stream-batch融合探索。最上面是视频、商业化等各种用户。在流批融合中，流的特点是低延迟，批的特点是高吞吐。对于流批一体化，我们希望系统不仅能处理非字段批数据，还能调整数据块的shufflesize，以平衡作业的吞吐量和延迟。快手内部对流批融合做了很多探索。我们建立了统一的Schema标准来存储数据，包括流表和批处理表。用户可以使用相同的代码来处理流表和批处理表，只是配置不同。生成的结果也需要符合统一的Schema标准，这样才能打通上下游，实现尽可能多的逻辑复用。Schema统一是我们快手数据治理的一部分，湖仓一体化等场景也有这个需求。应用场景主要包括以下几个方面：指标计算，比如实时指标和报表计算。数据回溯，利用已有的离线数据重新生成其他指标。数仓加速主要是数据仓库和数据湖的实时加速。流批融合带来的好处是多方面的。第一，降低开发和运维成本，实现尽可能多的代码逻辑复用，运维不再需要维护多个系统。其次，实时处理和批处理的口径一致，保证了最终结果的一致性。最后，还有资源优势。有些场景只需要一个实时系统。我们在调度方面进行了优化。对于上图所示的三个任务，最初a和c已经完成，b还在运行。这时候a失败了，按照默认的策略，需要重新运行ABC，即使c已经完成了。在实际场景中，会出现大量的c重新计算，造成巨大的资源消耗。在这种情况下，我们默认开启如下策略：如果a的结果是确定性的（其实大部分batch的输出都是确定性的），就可以不再重新计算c，只需要计算a和b即可。上图是我们内部针对批处理的优化和改进。首先是shuffle服务，现在有内部集成版和试用社区版，主要是在提升shuffle性能的同时实现存储和计算的解耦。二是动态资源的调度，主要是根据数据量自动判断算子的并发数，避免人工反复调整。三是慢节点回避，也叫推测执行，主要是为了减少长尾效应，减少总执行时间。第四是hive优化，比如UDF适配，语法兼容。另外，我们加入了缓存、多线程生成等方式来为分区生成拆分，大大减少了拆分的时间。最后支持一些压缩方式，比如gzip、zstd等。05未来规划我们未来的规划主要分为以下几个方面：第一，实时计算，进一步增强Flink的性能、稳定性和适用性，加速各种业务通过实时计算场景。二是线上线下统一，包括实时、准实时和批处理。我们期待通过Flink来统一快手数据同步、转换、离线计算，让ETL、数据仓库、数据湖处理等各种场景都可以使用一个Flink计算系统。最后一个是弹性伸缩，主要和云原生有关，包括离线混合和作业的弹性伸缩。
