flink metric 如何获取yarn applicationId？

要获取 Flink 在 YARN 集群上的 Application ID，可以通过 Flink 的 API 来获取，如 ClusterID 和 SessionID。
下面是两种常用的方法：

使用 Flink API 获取 Application ID

• 方法一：通过 ClusterID 获取

String clusterId = env.getClusterId();

• 方法二：通过 SessionID 获取

String sessionId = env.getSessionId();

使用 YARN REST API 获取 Application ID

除了使用 Flink 的 API 外，还可以使用 YARN 的 REST API 来获取 Application ID。首先，需要获取 ResourceManager 的 URL，并发送 GET 请求到 /ws/v1/cluster/apps 接口，如：

http://<ResourceManager>:8088/ws/v1/cluster/apps?state=RUNNING

然后，在响应中找到对应的 Application ID 即可。这种方法的优点在于无需在代码中添加额外的逻辑，但需要有一定的编程技巧和 YARN 集群的访问权限

在 Apache Flink 的 Metric 系统中，可以使用 YarnMetricQueryService 类来获取 Yarn Application ID。
首先，在 Flink 作业启动时，需要先创建一个 YarnMetricQueryService 实例，并将其添加到 Metric 注册表中。这样，Flink 作业就可以在运行过程中使用 YarnMetricQueryService 来查询 Yarn Application ID 以及其他 Yarn 相关的指标。
以下是一个示例代码片段，展示了如何创建并注册 YarnMetricQueryService：

// 创建一个 Flink Metric 注册表
MetricRegistry registry = new MetricRegistry();

// 创建一个 YarnMetricQueryService 实例
YarnMetricQueryService yarnMetricQueryService = new YarnMetricQueryService();

// 将 YarnMetricQueryService 添加到 Metric 注册表中
registry.register("yarn", yarnMetricQueryService);

// 使用 Metric 注册表创建一个 Flink Metric 温控器
MetricGroup metricGroup = new DefaultMetricGroup(registry);

// 在 Metric 温控器中获取 YarnApplicationID 指标
String yarnApplicationID = metricGroup.getIOMetricGroup().getCounter(YarnMetricQueryService.YARN_APPLICATION_ID);

在上面的代码中，YarnMetricQueryService.YARN_APPLICATION_ID 是一个常量，表示 Yarn Application ID 的指标名称。您可以使用这个名称从 Metric 温控器中获取 Yarn Application ID。
请注意，只有在 Flink 作业正在 Yarn 集群上运行时，才能使用

在 Flink 中，可以通过以下方式获取 Yarn applicationId：

1. 通过 org.apache.flink.runtime.yarn.YarnClusterDescriptor 获取 Yarn applicationId。

YarnClusterDescriptor yarnClusterDescriptor = (YarnClusterDescriptor) clusterDescriptor;
String yarnApplicationId = yarnClusterDescriptor.getApplicationId();

1. 通过 org.apache.flink.runtime.jobmaster.JobMasterGateway 获取 Yarn applicationId。

JobMasterGateway jobMasterGateway = ((JobManagerRunner) runner).getJobMasterGateway();
String yarnApplicationId = jobMasterGateway.getYarnApplicationId();

1. 通过 org.apache.flink.runtime.jobmaster.JobID 获取 Yarn applicationId。

JobID jobId = ((JobManagerRunner) runner).getJobID();
String yarnApplicationId = jobId.toString();

1.使用 YARN ApplicationStatus：

// 获取 MetricQueryService 对象
MetricQueryService metricQueryService = ...; // 通过 Flink 的 MetricQueryService 实例获取

// 获取 YARN Application ID
ApplicationId yarnApplicationId = metricQueryService.getApplicationStatus().getYarnApplicationId();

1. 使用 YARN ApplicationMaster 的环境变量：
   String yarnApplicationId = ParameterTool.fromSystemEnvironment().get("APPLICATION_ID");

By chatGPT 此回答整理自钉群“【②群】Apache Flink China社区”