FlinkForwardAsia2020：超大规模Flink调度优化

整理&＃xff1a;MWT&＃xff08;Apache Flink China 社区志愿者&＃xff09;
校对&＃xff1a;DJG&＃xff08;Apache Flink China 社区志愿者&＃xff09;

摘要&＃xff1a;本文主要介绍在字节跳动在比较大的规模的 Flink 应用场景下遇到的一些问题&＃xff0c;以及在调度层面所做的一些优化。内容主要包括6个方面&＃xff1a;

• 相关背景&＃xff1a;字节跳动的作业规模&＃xff0c;以及在此规模下遇到的问题与挑战&＃xff1b;
• 黑名单机制&＃xff1a;Flink 层面针对 NodeManager 异常节点所做的优化&＃xff1b;
• 启动速度优化&＃xff1a;Flink 层面针对 TaskManager 启动的长尾问题所做的优化&＃xff0c;以及 Yarn 层面针对 Container 启动速度的优化&＃xff1b;
• 调度器优化&＃xff1a;Yarn 层面针对流式场景所做的调度器的优化&＃xff1b;
• 资源隔离优化&＃xff1a;Yarn 层面针对 NUMA 架构和 CPU 争夺所做的优化&＃xff1b;
• 未来展望&＃xff1b;
  

1. 相关背景

1.1 业务规模

• 作业数量多
  字节跳动日常运行 10k&＃43; 个流式作业&＃xff0c;通过这些流式作业支持流式数仓、广告、机器学习、风控安全等业务。运行超过 80k&＃43; batch 作业&＃xff0c;主要支持数据通道任务。
• 机器规模大
  字节跳动的 Flink作业 运行在 Yarn 之上的&＃xff0c;由 Yarn 提供计算资源&＃xff0c;目前来看由 Yarn 提供的计算节点已经达到 10k&＃43;。
• 单作业并发大
  单个作业的 QPS 能超过 50M&＃43;&＃xff1b;算子并发能超过 30k&＃43;&＃xff0c;这样算下来&＃xff0c;整个拓扑图的边数量已经达到亿级别了。
  随着作业规模在不断增大&＃xff0c;并且单作业并行所需要的资源也在扩大&＃xff0c;也就带来了一些问题和挑战。

1.2 问题与挑战

• 机器规模大
• 用户希望更快的启动作业&＃xff0c;减少停流时间
  Flink 作为一个实时计算引擎&＃xff0c;它需要保证数据的时效性&＃xff0c;在作业启动或重启时&＃xff0c;能尽可能的快&＃xff0c;减少数据流的停顿时间。
• 机器 Quota 不均衡导致资源碎片
  在没有限制用户资源的情况下&＃xff0c;用户可以随意配置 CPU 、内存资源&＃xff0c;经过 Yarn 调度之后&＃xff0c;可能导致部分机器的 CPU 已经分光了&＃xff0c;但内存还剩余很多&＃xff0c;剩下的一部分机器内存已经分光了&＃xff0c;但 CPU 还剩余很多。由此&＃xff0c;整体表面上看是有资源的&＃xff0c;但这些资源完全分不出去。
• 实时作业对机器 load 敏感&＃xff0c;期望避开高 load 节点
  机器高 load 会影响实时作业的计算能力&＃xff0c;进而会导致数据积压&＃xff0c;数据产出时效就不能得到保证。所以&＃xff0c;希望任务调度 Container 或 TaskManager 时&＃xff0c;可以避开这些高 load 的机器节点。
• IO 密集型作业期望 Container 尽可能打散
  IO 密集型作业可能会频繁访问磁盘&＃xff0c;或者单机会有一个 Agent 去做流量转发之类的任务。此时&＃xff0c;如果大量IO 密集型的 Container 被调度到同一台机器上&＃xff0c;对磁盘的性能和 Agent 的压力会比较大。所以&＃xff0c;希望尽可能将这些 Container 打散调度到不同机器上&＃xff0c;以减少对单机的冲击。

3. 黑名单机制

NodeManager 节点经常会出现一些软硬件问题&＃xff0c;针对这种情况&＃xff0c;在 Flink 层面研发了黑名单机制。下面先介绍一下 NodeManager 节点上常见的问题。

2.1 常见问题

单机故障一般分为两种&＃xff0c;一种是基础硬件问题&＃xff0c;另一种是软件环境问题。

• 基础硬件问题
  包括磁盘、内存、CPU 等异常&＃xff0c;理论上这些异常应该由 Yarn 来提供保证。但是异常是非常多样的&＃xff0c;这种多样性导致我们不能 100% 快速检测定位到异常&＃xff0c;并处理对应的故障机器&＃xff0c;这就要求 Flink 层面能有效的兼容这些异常问题。
• 软件环境问题
  举个例子&＃xff0c;在字节跳动内部&＃xff0c;有部分数据没有直接写到 Kafka&＃xff0c;会先写到本地的 Agent&＃xff0c;再由 Agent 转发写到 Kafka。由于这些 Agent 并没有托管到 Yarn 上&＃xff0c;所以当 Agent 挂掉之后&＃xff0c;Yarn 并不能感知到这台机器的故障。Agent 挂掉会切实影响到 Flink 作业的运行。

2.2 异常节点处理逻辑

Flink 当前的异常节点处理逻辑如下&＃xff1a;

• Task 异常不会销毁 TaskManager &＃xff0c;重新调度后可能会再次失败
  假设当前有一个 Task 由于 Agent 异常或磁盘故障 fail 了&＃xff0c;fail 之后它会通知 JobManager , JobManger 根据恢复策略或重启策略进行一次故障。但是&＃xff0c;Task 的故障并不会引起 TaskManager 的故障&＃xff0c;当前的 TaskManager 还在正常提供服务。JobManager 在进行故障恢复时&＃xff0c;还会将出现故障的 Task 调度回之前的 TaskManager , 经过了一轮故障恢复之后&＃xff0c;故障并没有消失&＃xff0c;Task 在运行一段时间之后可能还会故障。循环往复的进行 Task 异常处理。
• TaskManager 异常后重新申请节点&＃xff0c;仍有可能分配到异常节点
  TaskManager 出现异常之后&＃xff0c;会由 Yarn 通知 JobManager Container 已经挂掉&＃xff0c;JobManager 会向 Yarn 申请一个新的 Container&＃xff0c;但此时 Yarn 可能并不知道之前挂掉的 Container 所在的节点已经故障了&＃xff0c;可能会将 TaskManager 重新调度回来。所以即使申请了新的 Container &＃xff0c;也不能解决异常节点问题。只能等 Yarn 随机调度到一个正常节点。
• Job 重新提交也有可能将 TaskManager 分配到异常节点
  加上了之前提到的 Quota 均衡、Container 打散等 feature 之后&＃xff0c;很有可能还会将 TaskManager 调度回同一异常节点&＃xff0c;这就导致作业一直在 failover 。

2.3 黑名单机制

针对上述现状研发了如下图所示的黑名单机制。

原理比较简单&＃xff0c;由 Flink 来感知节点异常&＃xff0c;当一个节点出现多次异常之后&＃xff0c;就不希望再将 Container 调度到这个节点上。黑名单机制有如下特点&＃xff1a;

1. 支持 Task 和 TaskManager 两个维度
2. 支持异常过滤&＃xff0c;减少误判&＃xff0c;例如 RemoteTransportException 等异常
   Task 异常上报中&＃xff0c;自定义 Task reporter 的 Exception Filter 来做一层异常处理拦截&＃xff0c;过滤一些已知的并不属于本机故障的异常&＃xff0c;例如 RemoteTransportException 异常&＃xff0c;明显是由于对端的异常导致连接异常&＃xff0c;不应该将其拉黑&＃xff1b;
3. 主动释放异常 TaskManager
   经过 Exception Filter 之后&＃xff0c;会将异常更新到 BlacklistTracker 里&＃xff0c;由 BlacklistTracker 来记录异常。当达到一定的阈值之后&＃xff0c;会主动在 BlacklistAction 中释放异常 Container 。
4. 异常节点上报给 Yarn
   向 Yarn 汇报更新黑名单&＃xff0c;Yarn 感知到之后会分配一个新的 Container&＃xff0c;就不会在这台机器上了。TaskManager 的异常处理过程和 Task 的异常处理过程类似&＃xff0c;先由 Yarn 汇报异常&＃xff0c;经过一些过滤操作之后&＃xff0c;再将黑名单更新回 Yarn。这里过滤的是常规的 Yarn 运维操作&＃xff0c;比如下机器等符合预期的操作。
5. 黑名单具有定时清理&＃xff0c;最大长度策略
   防止把线上的机器全拉黑导致作业无法正常运行。

2.4 上线收益

• 每天会有 100&＃43; 作业触发拉黑操作
• 触发拉黑节点的作业中 90% 拉黑了 20 台以下机器
  以上两点&＃xff0c;都符合业务预期。

3. 启动速度优化

3.1 常见用户反馈

• 作业启动时间不稳定
  作业启动涉及到很多的 IO 操作&＃xff0c;也会和很多外部系统交互&＃xff0c;比如向 HDFS 提交 jar 包&＃xff0c;由 Yarn 启动 Container&＃xff0c;下载 docker 镜像等等操作。由于这些外部交互&＃xff0c;导致了作业的启动可能是不稳定的。
• 作业启动时间靠经验和用户反馈来估计&＃xff0c;没有具体的指标
  用户希望平台能提供作业启动的指标&＃xff0c;但之前没有这个指标&＃xff0c;只能按照经验给出大概值&＃xff0c;小作业 1-2 分钟启动完毕&＃xff0c;大作业 7-8 分钟启动完毕。
• 作业实时性要求高&＃xff0c;要求重启时间尽可能短&＃xff0c;大作业尤其突出
  比如广告用户和直播榜单业务场景&＃xff0c;直面的是真实的用户&＃xff0c;关乎收入问题。所以对作业的实时性要求比较高&＃xff0c;推动了对于作业启动时间的优化。

3.2 Flink 作业启动流程

先分析一下作业启动流程&＃xff0c;涉及到 3 个 Flink 模块&＃xff0c;分别 Flink Client&＃xff0c;Flink JobManager 和 Flink TaskManager。

• Flink Client 会构建 StreamGraph、JobGraph&＃xff0c;完成之后会上传 jar 包到 HDFS&＃xff0c;并通知 Yarn 启动一个 ApplicationMaster。
• 在 ApplicationMaster 内部会启动 Flink JobManager&＃xff0c;会构建 ExecutionGraph 物理执行图。当物理执行图构建完成之后&＃xff0c;才会执行相关调度&＃xff0c;在调度过程中会触发 Container 的申请&＃xff0c;向 Yarn 申请 TaskManager。
• 等所有的 TaskManager 启动之后&＃xff0c;才会将提交的 Task 真正部署到指定的 TaskManager &＃xff0c;并且把作业拉起来。
  以上整个流程完成之后&＃xff0c;作业才算成功启动。
  

指标分析前期&＃xff0c;就是将上述的几个启动步骤依次拆解成上图的几个模块&＃xff0c;并进行相关指标数据打点&＃xff0c;输入到数仓中。

3.3 指标分析

下面对Flink作业的启动时间相关的统计信息进行分析。

下图为所有作业启动时间的分布。

下图是启动流程各阶段耗时在整体中的占比。

占比最大的是去申请资源&＃xff0c;也就是调度申请 Container 的步骤&＃xff0c;整个占比已经超过 50% &＃xff0c;所以 Container 启动算是一个比较慢的节点。占比第二大的是上传 jar 包到 HDFS。占比第三大的是启动 ApplicationMaster 。 由此看来&＃xff0c;启动 JobManager 和 TaskManager 耗时已经超过了作业启动时间的 80%。

根据两图可以得到以下结论&＃xff1a;

• 90% 的作业启动时间在 2min 以下&＃xff0c;启动时间比较长的作业占比还是比较少的&＃xff0c;但是有的作业启动时间会超过 5 分钟&＃xff0c;这是亟需优化的。
• TaskManager 启动时间占比超 50% &＃xff0c;
• JobManager 启动时间占比超 30% &＃xff0c;

下图为 TaskManager 启动耗时的分析。

可以得到以下结论&＃xff1a;

• 99.5% 的 TaskManager 能在 1min 内启动&＃xff0c;有 0.5% 超过了 1min&＃xff0c;甚至超过了 4min。
• Flink PerJob on Yarn 的作业需要等待所有的 TaskManager 启动&＃xff0c;根据木桶效应&＃xff0c;启动时间最长的 Container 就会影响作业的整体启动时间。

3.4 慢节点机制

针对某些 Container 启动时间过长的情况做了如下优化&＃xff1a;

1. 根据 TaskManager 启动情况&＃xff0c;动态推测慢节点阈值
   如申请 100 个 Container&＃xff0c;有 90 个Container 已经启动完成&＃xff0c;记录下当前时间&＃xff0c;乘以一定的系数。
2. 针对慢节点进行冗余申请 TaskManager &＃xff0c;以提升 TaskManager 启动速度
   认为超过阈值时间的节点就是慢节点&＃xff0c;就去冗余申请新的 TaskManager。
3. 已注册的 TaskManager 充足后及时释放冗余的节点&＃xff0c;减少资源占用
   当有足够多的 TaskManager 启动之后&＃xff0c;作业就正常恢复了&＃xff0c;需要释放掉之前冗余申请的 Container。达到和原生 Flink 一样按需申请的目的。

3.5 能不能更快

虽然慢节点机制解决了 TaskManager 启动长尾问题&＃xff0c;但是 JobManager 和 TaskManager 启动时间仍占 80% 。能不能更快一点&＃xff1f;那就需要对 JobManager 启动时长做进一步优化。

3.6 Session mode & Worker Pool

经过分析&＃xff0c;发现 JobManager 的启动中&＃xff0c;上传 jar 包到 HDFS &＃xff0c;构建 jar 包等等这些逻辑是比较慢的。于是&＃xff0c;想到了社区的 Session mode。Session mode 和 Per Job mode 的区别是&＃xff0c; Per Job mode 会为每一个作业提交一个 Yarn Application&＃xff0c;重复构建 Graph 、 构建 AM、申请 Container 等一系列流程。而 Session mode 在第一次向 Yarn 提交一个 Flink session 并启动一个 JobManager 之后&＃xff0c;这个 JobManager 是不会退出的&＃xff0c;省去了 JobManager 的启动时间。

• Per Job mode 的优点
  可以做到作业的资源隔离&＃xff0c;作业之间不会出现抢占资源的情况。
• Session Mode 的优点
  Session mode 由于省略了 AM 和 Container 的启动&＃xff0c;基本是可以将作业的重启时间控制在 30s 以内的。
  Session mode 并不是万能的&＃xff0c;有时需要更新作业的 JVM 参数&＃xff0c;或者修改 Flink Cluster 维度的动态参数&＃xff0c;都需要重启 Flink 的 JVM 进程&＃xff0c;销毁现有 Container 并重新申请 Container&＃xff0c;字节跳动 Yarn 团队针对这种场景开发了 work pool 组件来进行优化&＃xff0c;work pool 的详细介绍参见下文。

下面分析一下Yarn Session 模式下&＃xff0c;不同的重启场景是如何对重启时间进行优化的。

1. SQL 作业或者 DataStream Jar 包变更&＃xff0c; 不需要重启 Flink 集群的 JVM 进程&＃xff0c;借助 Session mode 可以将断流时间降至 30s 以下&＃xff1b;
2. 作业逻辑不变更&＃xff0c;但是想通过增加 TaskManager 的数量来实现动态扩容以增加Flink集群的消费能力。这种场景下可以先对 Flink 集群进行动态扩缩容&＃xff0c;然后在 Session mode 下重启作业。这里既能保证断流时间尽量短&＃xff0c;又能保证重启时 Flink 集群有足够的资源&＃xff1b;
3. 如果需要修改 Yarn Session 集群参数或者 Flink 版本&＃xff0c;无法使用 Session mode 来降低断流时间。这种场景考虑使用 worker pool 更新配置并原地重启 JVM 进程&＃xff0c;将预期断流时间减少到 1min 以内&＃xff1b;

4. 调度器优化

字节跳动的所有 Flink 作业都是跑在 Yarn 之上&＃xff0c;这里介绍一下 Yarn 团队在支撑流式场景方面碰到的一些问题以及挑战。第一点就是 Yarn 的调度器。

4.1 Fair Scheduler

熟悉 Hadoop 的同学都知道&＃xff0c;Fair Scheduler 是社区原创的调度器&＃xff0c;字节 Yarn 团队最开始是用 Fair Scheduler 来支撑流式场景&＃xff0c;但是碰到了比较多的问题。这里先介绍一下 Fair Scheduler 的机制。

图中最左边是 Flink 作业的 Application Master&＃xff0c;也就是我们平时说的 AM&＃xff0c;它是 Flink 作业的大脑&＃xff0c;等 AM 启动之后&＃xff0c;它会与 ResourceManager 维持一个定期的心跳来申请以及释放资源。ResourceManager 就是 Yarn 的大脑&＃xff0c;它管理所有的资源&＃xff0c;其中最核心的一块就是调度器&＃xff0c;例如 Fair Scheduler&＃xff0c;它的调度流程是这样的&＃xff1a;首先任务调度由 NodeManager 节点与 ResourceManager 的心跳触发&＃xff0c;也就是说每个 NodeManager 节点进行心跳的时候会触发一次调度&＃xff0c;Faire Scheduler 会将所有的 APP 进行一次排序&＃xff0c;找出最优的 APP 如图中的 APP2&＃xff0c;然后给选出的 APP 分配一个 Container。在下一次排序的过程中&＃xff0c;再给另外一个APP分配Container。

这个机制存在两个较大的问题&＃xff1a;

1. 第一个是资源互锁问题
   打个比方&＃xff0c;一个队列中持有1000个 vCore&＃xff0c;这时来了两个作业&＃xff0c;第一个作业需要1000个 vCore&＃xff0c;第二个作业也需要1000 vCore&＃xff0c;但是按照 Faire Scheduler 这种雨露均沾的分配模式&＃xff0c;最终的结果是每个作业只拿到一部分 vCore&＃xff0c;比如第一个作业只拿到了500个 vCore&＃xff0c;第二个作业也只拿到500个vCore&＃xff0c;每个作业都会因为拿不到完整的资源而出现 hang 住的情况&＃xff0c;我们称之为资源互锁问题。
2. APP无法主动选择NodeManager

刚刚提到&＃xff0c;调度是由 NodeManager 和 ResourceManager 的心跳触发&＃xff0c;每次由 NodeManager 选中一个 APP 来启动 Container&＃xff0c;但是对于 APP 而言&＃xff0c;总是被动地从 NodeManager 里面去接收资源&＃xff0c;没有主动选择 NodeManager 的权利&＃xff0c;即无法从集群中所有的 NodeManager中进行过滤&＃xff0c;找出自己觉得最优的 NodeManager 去起容器运行 APP&＃xff0c;这也导致 Fair Scheduler 无法完成将 Container 打散的需求。所谓 Container 打散&＃xff0c;这里打个比方&＃xff0c;一个 Flink 作业需要申请1000个 Container&＃xff0c;并且希望这1000个 Container 平铺在每个 NodeManager 之上&＃xff0c;现有的 Fair Scheduler 的机制无法满足这种需求。

4.2 Gang Scheduler

为了解决上面的一系列问题&＃xff0c;Yarn 团队自研了 Gang Scheduler&＃xff0c;如图中所示&＃xff0c;使用 Gang Scheduler 替换了社区的 Fair Scheduler 来支撑流式场景。下面说明一下 Gang Scheduler 和 Fair Scheduler 的不同之处。首先&＃xff0c;如图所示&＃xff0c;每次 Flink 作业请求来临时&＃xff0c;作业会从全局的 NodeManager 去选择节点&＃xff0c;共有如下步骤&＃xff1a;

1. Filter&＃xff1a;过滤掉某些负载比较高或者存在其他问题的 NodeManager&＃xff1b;
2. Score&＃xff1a;用户希望尽量将 APP 启动在某些得分更高的 NodeManager 上&＃xff0c;这里的“得分”是用户针对 NodeManager 自定义的一些约束&＃xff1b;
3. Assign&＃xff1a;批量分配 NodeManager 给作业&＃xff0c;如图中在node1上启动了Container1&＃xff0c;在node2上启动了Container2&＃xff1b;

回到刚刚的例子&＃xff0c;一个队列持有1000个 vCore&＃xff0c;两个作业都尝试申请1000个 vCore&＃xff0c;按照 Gange Scheduler 的机制&＃xff0c;最终的结果就是一个作业要么分配1000个 vCore&＃xff0c;要么一个 vCore 都不分&＃xff0c;即一个作业拿到全部资源并成功启动&＃xff0c;另一个作业等待资源。 通过这种方式解决了调度的互锁问题和 APP 无法主动选择 NodeManager 的问题。这里反转了调度的视角&＃xff0c;之前是由 NodeManager 来触发&＃xff0c;现在是由 APP 来触发&＃xff0c;这跟 k8s 的流程非常相似。

基于 Gang Scheduler 做到了 Container 打散的约束&＃xff0c;满足了 Flink 的场景需求。借助这个架构&＃xff0c;字节跳动 Yarn团队又做了更多的约束&＃xff0c;比如节点属性的约束&＃xff0c;高 load 跳过的约束&＃xff0c;quota 平均&＃xff0c;还有 GPU 亲和性等等。

这里简单介绍一下高 load 跳过&＃xff0c;因为节点的负载总是动态变化的&＃xff0c;Flink是对延迟非常敏感的作业类型&＃xff0c;它总是希望跳过那些负载比较高的节点。我们支持了两种模式来进行高load跳过&＃xff0c;
如果设置了强约束&＃xff0c;那么会在 filter 阶段强制跳过高 load 的节点&＃xff0c;只会在 load 较低的节点上去运行 APP。
如果设置了弱约束&＃xff0c;那么会优先在比较健康的节点上去分配 APP&＃xff0c;直到最后真的没有资源了&＃xff0c;才会去高 load 节点去分配 APP。

除了关注调度器的功能&＃xff0c;Yarn 团队还特别关注了调度器的性能&＃xff0c;经过持续不断的优化&＃xff0c;最终实现了分配一个 Container 只需要100微秒左右的时间&＃xff0c;也就是说新调度器的吞吐可以达到上万 Container 每秒。

4.3 Gang Scheduler 遗留问题

Gang Scheduler 满足了 Flink 的大部分调度需求&＃xff0c;但是上线之后&＃xff0c;又碰上了一些新的问题&＃xff0c;这里主要列举两点&＃xff1a;

1. 节点的 load 会动态变化
   第一点就是节点的 load 会动态变化&＃xff0c;因为 NodeManager 节点上布了很多的 agent&＃xff0c;这些 agent 资源没有受管控&＃xff0c;因此在业务高峰期的时候&＃xff0c;这些 agent 可能会占用比较多的资源&＃xff0c;会造成对应 NodeManager 节点上运行的 Container 受到影响。如何应对节点后期 load 变高的情况是个问题。
2. 磁盘故障
   另外一个问题是在超大规模集群下&＃xff0c;机器的故障尤其是磁盘故障变成一个不可忽视的问题。如果 Flink 作业运行在故障机器上&＃xff0c;如何去应对也是个问题。集群比较小的时候&＃xff0c;这个问题大家可以忽略&＃xff0c;或者是采用一些运维的手段把它解掉。但是如果节点特别多&＃xff0c;例如字节跳动好几万节点的情况下&＃xff0c;在如此庞大的基数下发生故障的节点也会非常多&＃xff0c;这会对 Flink 作业产生很大的干扰。

4.4 反调度器

为了解决刚刚提到的一系列问题&＃xff0c;字节跳动 Yarn 团队自研了反调度器。如图&＃xff0c;node1 被标红了&＃xff0c;它最开始确实是绿色的&＃xff0c;后期由于一些原因&＃xff0c;导致 node1 节点变得不健康&＃xff0c;因此把它标黄。为了解决这个问题&＃xff0c;在 ResourceManager 内部做了一个反调度器&＃xff0c;它会实时监测每个作业的请求。在图中的例子中&＃xff0c;反调度器发现 node1 的状态从低 load 变成高 load&＃xff0c;违背了用户最开始的约束。这个时候反调度器会进行一系列的数据采集和分析&＃xff0c;找出致使 node1 负载变高的 Container&＃xff0c;复用 ResourceManager 和 AM 之间的心跳&＃xff0c;将这个罪魁祸首的 Container 返回给 AM&＃xff0c;由 AM 决定要不要 kill 掉。如果决定要 kill&＃xff0c;AM 会通过 RPC 通知 node1 将该 Container kill 掉&＃xff0c;同时 node1 的负载会恢复正常。

之所以不直接 kill 高负载的 Container&＃xff0c;而是将 kill Container 的权利交还给了 Flink 的 AM&＃xff0c;是因为 Flink 的 AM 会有自己的一套逻辑来判断是否应该 kill 掉 Container&＃xff0c;例如当前处于业务高峰期&＃xff0c;不能随便将 Container kill 掉&＃xff0c;又或者 Flink AM 认为 kill 掉 Container 会给作业造成延时等等。

4.5 作业重启问题

上文中对 Flink 作业各个链路的优化讲的非常多。这里想要补充的一点是&＃xff0c;Flink 作业是一种对延迟非常敏感的作业类型&＃xff0c;作业重启又是时效性中非常关键的一环。
Flink 作业可能会因为失败或者升级进行重启&＃xff0c;在这些过程中也发现了比较多的问题&＃xff1a;

1. 作业重启的时候原先的资源被另外一个作业占用
   在 Yarn 的视角看来这是非常符合预期的现象&＃xff0c;因为 Yarn 队列中的任务是会动态地去抢资源的。但是在 Flink 业务方的视角看来&＃xff0c;这是不符合预期的。因为业务方仅仅想更新一下 Flink 作业&＃xff0c;结果有可能作业就起不来了&＃xff0c;这是不能被接受的。
2. 作业重启后被调度到新的 NodeManager 节点上&＃xff0c;需要重新拉取 image
   如果作业重启后被调度到新的 NodeManager 节点上&＃xff0c;新的节点中是没有 Flink 作业原来的 image&＃xff0c;这个时候会进行重新拉取&＃xff0c;重新拉取如果时间过长&＃xff0c;也会造成 Container 启动过慢&＃xff0c;这也是用户吐槽比较多的一点。

4.6 Worker Pool

为了解决 Flink 作业重启失败或者重启慢的问题&＃xff0c;研发了一个独立的 service 叫做 work pool。

大家都知道 Yarn Client 中有一个 submitApplication 接口和 killApplication 接口&＃xff0c;分别用来提交作业和 kill 作业 &＃xff0c;我们并列地增加了一个 updateApplication 接口。

如上图所示&＃xff0c;ResourceManager 在收到 updateApplication 请求后&＃xff0c;会把原来的 AM1 删掉&＃xff0c;然后并且原地启动一个 AM2&＃xff0c;AM2 启动后跟 ResourceManager 之间建立心跳&＃xff0c;通过心跳来申请资源。AM2 在向 ResourceManager 申请资源时&＃xff0c;Worker Pool 会将所重启作业原先位于node1上的 Container1 和位于 node2 上的 Container2 原地转换成 Container3 和 Container4 并返回给 AM2&＃xff0c;由于并没有真正进行调度&＃xff0c;而是只是进行 Container 的转换&＃xff0c;因此重启时间会非常快。AM2 得到 Container3 和 Container4 之后&＃xff0c;会在 NodeManager 中将Container1和Container2进行原地升级。

纵观整个过程的&＃xff0c;暴露一个 updateApplication 接口&＃xff0c;使得重启后的作业使用上一次 attempt 的资源&＃xff0c;并没有发生真正的调度&＃xff0c;作业的资源自然而然不会被其他的作业抢过去。另外一点&＃xff0c;该方式尽可能去保证 Container 在原地进行重启来复用作业上一次 attempt 的 image&＃xff0c;使得 Container 拉起的时间非常短。

5. 资源隔离优化

下面介绍 Yarn 团队在资源隔离方面做得一些优化。

5.1 更丰富的隔离策略

首先介绍一下当前的 CPU 隔离机制&＃xff0c;以及存在的问题。

社区版的 Yarn 在拉起一个至少需要一个 vCore 的 Container 时&＃xff0c;会通过 CGroup 来保证该Container最少能用到一个 vCore&＃xff0c;但其实该 Container 最多可能会用到整机中所有的 vCore&＃xff0c;这种机制在批处理框架如 MapReduce 或 Spark 中是适用的&＃xff0c;因为它可以很好地提高单机的利用率&＃xff0c;但在流式场景下&＃xff0c;这种机制经常会造成 Flink 任务的延迟&＃xff0c;因为如果任务运行在 vCore 争抢中占下风的 Container中&＃xff0c;会造成 Flink 作业处理速度过慢的情况。

字节跳动 Yarn 团队对 Container 的 CPU 申请设置了上限&＃xff0c;在保证 Container 最少能获取到 n 个 vCore 的同时&＃xff0c;约束 Container 最多能获取 1.2*n 个 vCore。通过这种方式实现了不同 Container 对 CPU 的争抢问题。

另外&＃xff0c;对 NUMA 做了优化。单机很多都是 NUMA 结构&＃xff0c;如图所示存在两个 NUMA&＃xff0c;NUMA1 和 NUMA2。如果存在多 NUMA&＃xff0c;尽可能把一个作业分配在同一个 NUMA 上。如图所示作业1的 Container1 和 Container2 被分配在了 NUMA1&＃xff0c;通过 NUMA 天生隔离的方式&＃xff0c;实现了不同作业更好的隔离特性。

5.2 磁盘 IO 瓶颈问题

Flink 经常有很多 checkpoint 的需求&＃xff0c;如果机器磁盘的 IOPS 性能不高&＃xff0c;会造成 checkpoint 过慢甚至失败的情况。由于历史原因&＃xff0c;很多机器磁盘都是 HDD&＃xff0c;而经过测试发现&＃xff0c;HDD 的 IOPS 只有100左右。为了解决这个问题引入了远程盘&＃xff0c;测试的结果是它比老的 HDD 的 IOPS 提升大约25倍左右。远程盘虽然有网络的开销&＃xff0c;但它底层使用了更好的硬件例如 SSD 或 NVMe 等。使用远程盘之后 checkpoint 失败的情况没有再发生&＃xff0c;且 checkpoint 速度得到了极大提升。

6. 未来展望

接下来介绍一下未来的展望&＃xff0c;分成 Flink 和 Yarn 两个环节。

1. Flink层面&＃xff1a;

• 进一步优化作业启动、恢复速度&＃xff0c;这是用户特别关心的&＃xff1b;
• 更好地支持 OLAP 场景&＃xff1b;
• 探索 Auto Scaling&＃xff1b;

2. Yarn层面&＃xff1a;

• 更丰富的调度策略
  随着 Flink 的广泛应用&＃xff0c;用户会提出越来越多的调度需求&＃xff0c;为了支撑用户&＃xff0c;会去研发不同的调度策略甚至是调度器来进行支持&＃xff1b;
• 调度可解释化
  虽然自研了新的调度器&＃xff0c;并做了很多约束&＃xff0c;研发了很多功能来支撑用户&＃xff0c;但是为了能向用户解释调度的结果真的符合预期&＃xff0c;真的准确&＃xff0c;还需要一个可解释化的方案&＃xff1b;
• 探索 Container 热迁移
  现在的反调度策略是把高负载的 Container 杀掉&＃xff0c;然后来保证 NodeManager 节点的健康&＃xff0c;但带来的代价就是 Flink 作业也会被杀掉。正在探索一种更高级的方法&＃xff0c;把受影响的 Container 迁移到新的节点&＃xff0c;并恢复该 Container 原先的状态。