Flink on YARN新版本如何指定Task Manager数量？

在Flink 1.12及后续版本中，原有的-n（会话模式）和-yn（Per-Job模式）参数已被移除，取而代之的是通过配置属性或命令行参数来管理资源。以下是具体的配置方法和核心资源属性说明：

一、设置Task Manager数量的方式 #

1. 会话模式（Session Mode） #

会话模式下，你可以通过两种方式指定Task Manager数量：

• 命令行参数：启动YARN会话时，通过-D参数传递配置，示例：

  ./bin/yarn-session.sh -Dtaskmanager.numberOfTaskManagers=3
  

• 配置文件：在flink-conf.yaml中添加或修改以下配置，启动会话时会自动读取：

  taskmanager.numberOfTaskManagers: 3
  

  注意：命令行参数的优先级高于配置文件，若两者同时设置，以命令行参数为准。

2. Per-Job模式（Per-Job Mode） #

Per-Job模式下，推荐通过命令行参数为单个作业指定资源，也可通过配置文件全局设置：

• 命令行参数：提交作业时附带-D参数，示例：

  ./bin/flink run -m yarn-cluster -Dtaskmanager.numberOfTaskManagers=2 ./examples/streaming/WordCount.jar
  

• 配置文件：在flink-conf.yaml中设置taskmanager.numberOfTaskManagers，该配置会作用于所有Per-Job提交的作业。

如果开启了动态资源分配（默认启用），Flink会根据作业的运行负载自动调整Task Manager数量，此时taskmanager.numberOfTaskManagers将作为初始数量，最终数量会在配置的最小/最大值范围内波动。

二、核心资源配置属性 #

以下是控制Flink在YARN上资源分配的关键配置，可在flink-conf.yaml中全局配置，或通过启动命令的-D参数临时指定：

Task Manager资源配置 #

• taskmanager.numberOfTaskManagers: 固定Task Manager数量，动态分配场景下作为初始值
• taskmanager.memory.process.size: Task Manager进程的总内存（含JVM堆、堆外内存等），格式如4096m
• taskmanager.numberOfTaskSlots: 每个Task Manager的任务槽数量，默认值为1
• yarn.taskmanager.resource.cpu-vcores: 单个Task Manager占用的CPU核数，默认值为1

YARN应用全局配置 #

• yarn.application.name: YARN上Flink应用的名称，便于在YARN UI中识别
• yarn.application.queue: 作业提交到YARN的指定队列，默认使用YARN的默认队列
• yarn.resourcemanager.am.resource.memory-mb: Application Master的内存大小
• yarn.resourcemanager.am.resource.cpu-vcores: Application Master占用的CPU核数

动态资源分配配置 #

• dynamic.resource.allocation.enabled: 是否启用动态资源分配，默认值true
• dynamic.resource.allocation.min-taskmanagers: 动态分配时允许的最小Task Manager数量
• dynamic.resource.allocation.max-taskmanagers: 动态分配时允许的最大Task Manager数量
• dynamic.resource.allocation.initial-taskmanagers: 动态分配时的初始Task Manager数量

内容的提问来源于stack exchange，提问作者Dagang Wei