1、冯斐/王非凡 美团数据平台工程师美团Flink大作业部署与状态稳定性优化实践相关背景#1大作业部署优化#2Checkpoint跨机房副本#3未来规划#5目录状态稳定性相关优化#42#1相关背景3美团Flink应用场景数据管道数据分析事件处理-数据实时入仓-跨数据源实时同步-实时数仓-实时特征生产-安全风控-系统监控告警4机器数美团Flink计算规模15K50K 5.4亿/秒 作业数5K高峰流量作业最大并发10TB作业最大状态5面临的问题与挑战大规模作业部署运行问题-JobManager部署Task过程耗时久，或出现RPC超时，导致作业启动失败-部分TaskManager上Task较多，导致Ta

2、sk启动失败-大作业制作Checkpoint给HDFS带来压力，影响其他作业稳定运行状态稳定性和效率问题-大状态作业在制作Savepoint期间，CPU和内存开销高，影响作业正常运行-基于Savepoint的恢复，需要回溯的数据多，作业恢复效率低、耗时长-高优先级作业的Checkpoint有机房级别的容灾需求6#2大作业部署优化7作业规模及典型问题-算子最大并发达到5K，-Task总数8K，2层shuffle，1个region-TaskManager数量1K+大作业的规模-JobManager部署大量Task过程耗时久，或者出现RPC超时，导致作业启动失败-部分TaskManager上Task

3、较多，申请不到足够的Network Buffer，导致作业启动失败-大作业制作Checkpoint给HDFS带来压力，其他使用HDFS的作业遇到Server too busy异常大作业的问题YARNHDFSSession ModeCheckpoint8JobManager部署作业流程分析构建执图申请资源部署Task启动Task 作业规模 作业拓扑复杂度 资源需求量 资源健康度 资源调度策略 资源调度性能 Task数量 TaskManager数量 作业拓扑复杂度 User Jar大小部署主要步骤部署影响因素当前规模下，问题集中在这两个步骤部署主要问题部署失败部署慢9分发User Jar问题分析现

4、象-指标：JobManager所在机器的网卡打满-日志：TaskManager下载User Jar耗时较多-测试：相同规模的作业缩小User Jar后，RPC超时异常消失，部署耗时减少HDFSBlobServerJobManagerBlobCacheTaskManagerTaskTaskJarBlobCacheTaskManagerTaskTaskJar由JM分发UserJar(关闭HA)由HDFS分发UserJar(开启HA)JM大量分发UserJar时，网络传输成为瓶颈每个TM都需要下载UserJar分析10分发User Jar问题优化HDFSBlobServerJobManagerBlo

5、bCacheTaskManagerTaskTaskJarBlobCacheTaskManagerTaskTaskJar每个TM都需要下载UserJarJM分发压力大，成为瓶颈HDFSBlobServerJobManagerBlobCacheTaskManagerTaskTaskJarBlobCacheTaskManagerTaskTask每个节点只下载1次UserJarJM分发压力减小，不再成为瓶颈每个节点上只下载1次User Jar，下载次数由TM粒度减小至机器粒度，降低了1个数量级11RPC超时问题分析与优化现象-有shuffle且并发度较大的作业仍然有updateTaskExecutio

6、nState等RPC超时的情况-JobManager侧存在大量requestPartitionState的RPC请求分析与优化上游Task(未启动)下游Task（已启动）1.requestPartitionsJobMaster2.requestPartitionState上游Task（未启动）下游Task（已启动）1.requestPartitions(with retry)JobMaster2.requestPartitionState下游Task先重试，减少请求JobMaster次数12优化效果-JobManager分发UserJar压力降低，作业部署耗时大幅减小，作业规模越大效果越明显-

7、当前大作业规模下，大作业部署过程RPC超时异常消失，大作业可成功部署User Jar 分发优化效果对比（测试多region作业）作业部署耗时100s200s300s400s作业TM数量060001000052s44s40s42s39s319s175s123s71s52s优化前优化后13Task分布不均问题分析问题-资源浪费：不同TaskManager实际需要资源量不同，但是都需要按照最大资源需求去申请资源-计算热点：不同TaskManager中Task数量分布情况不同，Task多的TaskManager更可能成为计算热点和计算瓶颈现象-Task分布较多的TaskManag

8、er不能提供足够的Network Buffer，Task启动失败-虽然可以通过增加TaskManager总内存、调整内存比例临时解决，但治标不治本14Task分布不均问题分类Task数量分布不均：不同算子的Task集中在部分TaskManagerSlotSource(1/2)Process(1/4)Sink(1/2)TaskManagerSlotProcess(2/4)SlotSource(2/2)Process(3/4)Sink(2/2)TaskManagerSlotProcess(4/4)Task类型分布不均：相同算子的不同Task集中在部分TaskManagerSlotSourceB(1

9、/1)Map(1/4)Sink(1/1)TaskManagerSourceA(1/1)SlotMap(2/4)SlotMap(3/4)TaskManagerSlotMap(4/4)SourceC(1/1)SourceD(1/1)15Task数量分布不均分析-原因分析：SlotSharing机制允许不同算子的不同Task使用同一个Slot，但在Task选择所属Slot时，未考虑Slot中Task的数量分布情况，导致多个Task可能集中在部分Slot上，进而造成Task集中在部分TaskManager中。-典型场景：多Source、多Sink等流量的汇聚分发作业SlotSourceB(1/1)Ma

10、p(1/4)Sink(1/1)TaskManagerSourceA(1/1)SlotMap(2/4)SlotMap(3/4)TaskManagerSlotMap(4/4)SourceC(1/1)SourceD(1/1)16Task数量分布不均优化SlotSourceB(1/1)Map(1/4)Sink(1/1)TaskManagerSourceA(1/1)SlotMap(2/4)SlotMap(3/4)TaskManagerSlotMap(4/4)SourceC(1/1)SourceD(1/1)Task选择Slot策略优化：-无上游的Task尽量使用新的Slot，直至Slot数到达上限-有上游

11、的Task优先选择上游Task所在的Slot，减少不必要的数据分发-有多个可选Slot时，选择Task数量少的Slot，使Task数量尽量均匀SlotMap(1/4)Sink(1/1)TaskManagerSourceA(1/1)SlotMap(2/4)SlotMap(3/4)TaskManagerSlotMap(4/4)SourceB(1/1)SourceC(1/1)SourceD(1/1)17Task类型分布不均分析-原因分析：一个Slot里不会出现多个相同类型的Task，只有包含相同类型Task的多个Slot集中在某个TaskManager时，才会导致Task类型集中；Slot选择Tas

12、kManager受Slot的申请顺序影响，但当前实现里Slot申请顺序是随机的，未考虑到Slot中Task类型的分布情况-典型场景：比较普遍，不同算子并发度不一致的作业SlotSource(1/2)Process(1/4)Sink(1/2)TaskManagerSlotProcess(2/4)SlotSource(2/2)Process(3/4)Sink(2/2)TaskManagerSlotProcess(4/4)18Task类型分布不均优化Slot选择TaskManager策略优化：按照Slot中Task的类型组合情况对Slot申请顺序进行调整，使包含相同Task类型组合的Slot尽量分布

13、到不同的TaskManager中SlotSource(1/2)Process(1/4)Sink(1/2)TaskManagerSlotProcess(2/4)SlotSource(2/2)Process(3/4)Sink(2/2)TaskManagerSlotProcess(4/4)SlotSource(1/2)Process(1/4)Sink(1/2)TaskManagerSlotProcess(3/4)SlotSource(2/2)Process(2/4)Sink(2/2)TaskManagerSlotProcess(4/4)19Task分布不均优化策略组合SlotSource(1/2)P

14、rocess(1/4)Sink(1/2)SlotProcess(2/4)SlotSource(2/2)Process(3/4)Sink(2/2)SlotProcess(4/4)20SlotSource(1/2)Process(1/4)Sink(1/2)SlotProcess(3/4)SlotSource(2/2)Process(2/4)Sink(2/2)SlotProcess(4/4)SlotSource(1/2)Process(1/4)SlotProcess(2/4)SlotSource(2/2)Process(3/4)Sink(2/2)SlotProcess(4/4)Sink(1/2)Ta

15、sk类型均匀分布Task数量均匀分布Task分布不均优化效果TM中Task数4681012TM个数391491152597优化前优化后Task的分布情况TM的CPU使用率TM的剩余Network Buffer个数Task的分布情况TM的CPU使用率TM的剩余Network Buffer个数21TM中Task数4681012TM个数01200000HDFS压力问题问题-随着业务增长，HDFS集群的负载日益增长，逐渐收到RPC请求量、排队时长告警-大作业制作Checkpoint时，HDFS CallQueue打满，影响其他作业读写HDFSUnit:1,000,000 US$0k200k400k60

16、0k800k---08710k609k430k358k240k80knamenode每分钟RPC请求量22HDFS压力应对/flink-nn01/.flink/checkpoints/savepoints/recoverynameservice1HDFS Cluster-state.checkpoints.dir-state.savepoints.dir-yarn.staging-directory-high-availability.storageDir任务均衡策略-考虑namenode负载-考虑业务分布-作业粒度指定

17、namenode水平拓展/flink-nn02/.flink/checkpoints/savepoints/recoverynameservice2nameserviceN动态指定路径-HDFS服务能力可扩展和应用-大作业的运行不影响其他作业最终效果23大作业部署其他优化-个性化调优：给用户开放Flink运行参数，用户可根据作业自身运行特点，进行个性化调优-异常作业拦截限制Checkpoint最小制作间隔，避免不合理的高频Checkpoint制作影响集群上其他作业24#3Checkpoint 跨机房副本25Checkpoint 为什么需要跨机房副本Flink 计算资源多机房交付，作业有换机房启

18、动的场景 1先前的 Savepoint 自动跨机房备份已经不能满足需求 4关键作业要求机房级别的状态容灾能力 3基于过往经验，更倾向于使用 Retained Checkpoint 而不是 Savepoint 恢复作业 226所有作业支持换机房从 Checkpoint 启动（启动前副本制作，原机房故障不可恢复）1关键作业的 Checkpoint 支持跨机房容灾（实时副本制作，原机房故障可恢复）2目标 27所有作业支持换机房从 Checkpoint 启动（启动前副本制作，原机房故障不可恢复）1关键作业的 Checkpoint 支持跨机房容灾（实时副本制作，原机房故障可恢复）2Checkpoint

19、Self-contained&Relocatable（自包含&可移动）Checkpoint Replicate Service（副本制作服务）目标&解决思路28/checkpoints/$job-idchk-3taskownedshared_metadataCheckpoint 目录结构29独有的状态文件checkpoint间共享的状态文件每个 checkpoint 的元数据文件存放永远不能由 jm 删除的文件存放各 checkpoint 之间共享的文件存放每个 checkpoint 独有的文件/checkpoints/$job-idchk-3taskownedshared_metadataC

20、heckpoint 目录结构30/checkpoints/$job-idchk-3taskownedshared_metadataCheckpoint 目录结构31/checkpoints/$job-id-xchk-1taskownedshared_metadata/checkpoints/$job-idchk-3taskownedshared_metadataCheckpoint 为什么不是 Self-contained32/checkpoints/job-3/checkpoints/job-1/checkpoints/job-2Checkpoint 为什么不是 Self-contained

21、job-1、job-2、job-3 为同一作业的多次启动33导致 Retained Checkpoint 难以清理；清理一个作业的 retained checkpoint 时要确保其中的文件不再被引用；因而需要在平台上维护 checkpoint 的引用计数，增加平台管理成本。1非 Self-contained 带来的问题导致跨存储系统的 Checkpoint 副本不可用；例如将 job_1 的 cp 从 HDFS_1 复制到 HDFS_2 上之后，由于其跨 job 实例引用的文件在 HDFS_2 上不存在，导致这个复制的 cp 不可用。234/dbmeta-files01.sst02.sst0

22、3.sstCheckpoint 如何实现 Self-contained35rocksdb 的数据文件，只读文件rocksdb 的元数据文件/dbmeta-files01.sst02.sst03.sstCheckpoint 如何实现 Self-contained36previous-sst-list/dbmeta-files02.sst03.sst04.sstdb status of checkpoint-5db status of checkpoint-3/dbmeta-files01.sst02.sst03.sst无需上传Checkpoint 如何实现 Self-contained37pre

23、vious-sst-list恢复构造 rocksdb 实例Checkpoint 如何实现 Self-contained38同一job？YNCheckpoint 如何实现 Self-contained39previous-sst-list恢复构造 rocksdb 实例Checkpoint 如何实现 Self-contained40Checkpoint 如何实现 Self-contained41为什么 Checkpoint 不是 Relocatable/cp/$job-id/chk-5/file-1/cp/$job-id/shared/file-2/cp/$job-idchk-5taskowned

24、shared42/cp/$job-id/chk-5/file-1/cp/$job-id/shared/file-2baseDir:/cp/$job-id/chk-5/./file-1./shared/file-2Checkpoint 如何实现 Relocatable43副本制作的备选方案使用 DistCp 对 Checkpoint 目录进行跨机房复制 1Flink 引擎制作 Checkpoint 时直接双写两个集群 3新增 CheckpointReplicateService 从外部进行 Checkpoint 跨机房复制 2制作 Checkpoint 之后 Coordinator 触发 Dis

25、tCp 对 cp 进行跨机房复制 444Checkpoint Replicate ServiceReplicate Servicehdfs-clienthdfs-client45Checkpoint Replicate ServiceReplicate Service/dir1/cp/job-1/chk-5referencedFiles-5/dir2/cp/job-1/chk-3referencedFiles-3_metadata of chk-5_metadata of chk-3files to replicate46Checkpoint Replicate ServiceReplicat

26、e Servicefiles-to-copyfiles-to-deleteignorereferencedFiles-5referencedFiles-347Checkpoint Replicate ServiceReplicate Service引擎 _metadata 解析的逻辑需要改写；当前实现会在解析过程中访问元文件所在的存储（HDFS），可能因不是服务默认连接的hdfs集群而解析失败。1考虑缓存 _metadata 解析结果；我们生产上一些大状态作业，_metadata 有几十M（甚至几G），引用文件超数十万个，解析时间能到分钟级。2引用文件的复制和删除拆分成子任务，多个节点并行执行

27、；大状态作业一次cp副本复制的数据量可能达到10TB+，单个节点网络容易到达瓶颈。3运行中作业的副本制作失败不做重试（实效性）；files-to-delete 执行失败不用失败整个 cp 副本任务（只是多一些残留文件，不影响checkpoint可用）；4一些工程经验：48#4状态稳定性相关其他优化49RocksDBStateBackend 内存泄漏触发条件：作业发生 restart，restart 之后会复用没有退出的 TM；TM heap 内存充足，full gc 不频繁；问题原因：RocksDBStateBackend 的清理过程存在 bug，有一处 RocksObject 没有清理；修复

28、版本：1.12.3 ISSUE ：https:/issues.apache.org/jira/browse/FLINK-21986 50Savepoint 导致增量 Checkpoint 退化触发条件：成功的 Savepoint 之后的第一个增量 Checkpoint 会退化成全量（上传所有rocksdb文件）；问题原因：Savepoint 制作完成后错误的清理了 previous-sst-list；修复版本：1.12 ISSUE ：https:/issues.apache.org/jira/browse/FLINK-23949 51触发时指定 Savepoint 超时时间大状态作业 Savepoint 制作时间一般远超 Checkpoint，但却直接采用后者的超时时间配置，导致需要给 Checkpoint 配置一个能够覆盖 Savepoint 的超时时间；问 题：修复版本：内部版本已经上线，社区版本PR还未完成；ISSUE ：https:/issues.apache.org/jira/browse/FLINK-946552#5 未来规划53未来规划稳定性建设-作业断流恢复时间优化-探索Flink on k8s资源效率提升-资源效率评估与优化运行性能优化-状态后端性能提升-反压优化54THANKS