Flink Logging 获取作业名称或作业 ID

我正在尝试设置 logback.xml，以便它将包含与日志记录关联的 JobName 或 JobId。

我还没有找到一种方法来做到这一点。是否可以？

最终我想要实现的是能够将日志发送到 ElasticSearch 并用消息标记 JobName 或 Id，以便我能够按作业进行过滤。

我们最终做了以下事情，感觉超级黑客我们希望能找到更好的方法。

1. 设置集群以使用 logback

2. 通过实现创建自定义 logback ContextSelectorch.qos.logback.classic.selector.ContextSelector

3. 将 jar 放在每个 Flink TaskManagers 类路径中

4. Modify FLINK_ENV_JAVA_OPTS_JM在每个 flink TaskManager 上指定您的上下文选择器。例如：

   导出 FLINK_ENV_JAVA_OPTS_JM="-Dlogback.ContextSelector=your.custom.context.Selector"

5. 重新启动所有任务管理器

6. Test it.

这是我们创建的 ContextSelector 的要点。如果您有任何改进，我很乐意提出改进请求！自定义 Flink Logback ContextSelector https://gist.github.com/Astn/a02baea87b33f213d0be873879c1a4e2#file-flinklogbackcontextselector-scala

这会为每个 JobID 创建一个 LoggerContext，然后使用 JobID 向 JobManager 发出请求以获取 JobName。然后，它用“:”分割 JobName，并将头组件放入 LoggerContext 属性“flink_job_name”中，并将尾组件放入 LoggerContext 属性“flink_job_version”中。