如何通过代码获取Flink Job Manager Web IP并实现作业提交自动化？

我正尝试通过流水线调用Flink REST API实现Flink作业提交流程自动化，调用这些接口需要知道Job Manager Web界面的IP。在POC阶段我是通过CLI执行flink-yarn-session命令手动获取该IP的，但现在想知道如何通过代码自动获取这个IP？

我计划按以下步骤调用REST API完成自动化：

• 调用GET http://<JobManager-IP>:8081/jobs/overview：获取运行中的作业ID
• 调用POST http://<JobManager-IP>:8081/jobs/:jobID/savepoints/：触发作业保存点并取消原作业
• 调用GET http://<JobManager-IP>:8081/jobs/:JobId/savepoints/:savepointId：查询保存点状态
• 调用POST http://<JobManager-IP>:8081/jars/upload：上传新作业的Jar包
• 调用POST http://<JobManager-IP>:8081/jars/<jar-id>/run：提交新作业（示例中Jar ID为de05ced9-03b7-4f8a-bff9-4d26542c853f_ATVPlaybackStateMachineFlinkJob-1.0-super-2.3.3.jar）
• 调用GET http://<新JobManager-IP>:<端口>/jobs/:jobId：获取新作业的运行状态

代码获取JobManager Web IP的可行方案 #

方案1：通过YARN REST API查询 #

Flink YARN Session作为YARN集群上的应用，可通过YARN的REST接口直接获取其JobManager地址：

1. 调用YARN资源管理器的应用列表接口，过滤出运行中的Flink应用：

GET http://<YARN-RM-IP>:8088/ws/v1/cluster/apps?applicationTypes=Apache Flink&applicationStates=RUNNING

2. 从返回结果中提取目标Flink应用的trackingUrl字段，该字段即为JobManager的Web地址，解析后可得到IP和端口。

Python代码示例：

import requests
from urllib.parse import urlparse

# 替换为你的YARN ResourceManager地址
yarn_rm_url = "http://<YARN-RM-IP>:8088/ws/v1/cluster/apps"
params = {
    "applicationTypes": "Apache Flink",
    "applicationStates": "RUNNING"
}

response = requests.get(yarn_rm_url, params=params)
apps = response.json()["apps"]["app"]

# 根据应用名称过滤目标Flink Session（可根据实际场景调整过滤条件）
flink_session = next(app for app in apps if app["name"] == "Flink session cluster")
job_manager_url = flink_session["trackingUrl"]

# 解析IP和端口
parsed_url = urlparse(job_manager_url)
job_manager_ip = parsed_url.hostname
job_manager_port = parsed_url.port

方案2：解析flink-yarn-session命令输出 #

如果你的流水线是通过代码启动Flink YARN Session，可直接捕获命令输出并提取JobManager地址：

Python代码示例：

import subprocess
import re
from urllib.parse import urlparse

# 启动Flink YARN Session（根据实际需求调整命令参数）
proc = subprocess.Popen(
    ["flink", "yarn-session", "-d"],
    stdout=subprocess.PIPE,
    stderr=subprocess.STDOUT,
    text=True
)

output, _ = proc.communicate()
# 匹配输出中的JobManager Web地址
match = re.search(r"JobManager Web Interface: (http://.*?:\d+)", output)
if match:
    job_manager_url = match.group(1)
    parsed_url = urlparse(job_manager_url)
    job_manager_ip = parsed_url.hostname
    job_manager_port = parsed_url.port

方案3：读取Flink临时文件（备选） #

Flink YARN Session启动后，会在本地~/.flink/application_<app-id>目录下生成日志或配置文件，其中可能包含JobManager地址。但该方法依赖本地文件结构，跨环境兼容性较差，仅作为备选方案。

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