Flink 1.11版本基于Kubernetes的运行集群如何正确进行扩缩容？

核心原因 #

Flink 1.11.x 版本原生不支持运行时作业的动态资源调整，作业启动时JobManager就已经根据当时的可用Slot数量、作业并行度配置固定了所有任务的Slot分配规则，运行过程中新增的TaskManager的空闲Slot不会被正在运行的作业自动感知和调度。

适配TM数量增减需要执行的操作 #

• 先调整作业并行度相关配置
  不要在作业代码中硬写固定并行度（比如调用env.setParallelism(xx)写死数值），也不要在提交作业时指定固定的-p并行度参数。提前设置合理的作业最大并行度（最大并行度一旦设定无法修改，需要提前预留足够的扩容余量），给后续并行度调整留空间。
• 手动触发作业带保存点重启
  1. 为运行中的作业创建保存点：
     flink savepoint <作业ID> <分布式存储路径（如hdfs、s3路径）>
  2. 停止当前运行的作业：
     flink cancel <作业ID>
  3. 从生成的保存点重启作业：
     flink run -s <上一步生成的保存点全路径> <作业Jar包参数>
     重启后JobManager会重新扫描集群所有可用的TM Slot，按照最新的资源情况重新分配任务，新增TM的空闲Slot就会被使用。
• 半自动化扩缩容适配方案
  可以自行编写外部脚本监听Kubernetes TM Deployment的副本数变化，副本数调整完成后自动执行上述的保存点重启流程，不需要人工介入。需要注意该模式下作业会有短暂中断，中断时长取决于作业状态大小。
• 长期优化方案
  升级Flink集群到1.13及以上版本，1.13版本引入的自适应调度器可以自动感知集群空闲Slot，运行过程中自动调整作业并行度、重平衡任务分配，扩缩容TM后不需要手动重启作业即可自动适配。

额外说明：如果你用的是Flink Session集群，新增TM的Slot默认会优先分配给新提交的作业，已经在运行的旧作业还是需要重启才能使用新Slot。

内容的提问来源于stack exchange，提问作者Kumar Padhy