将子任务id映射到Flink中的TaskManager

Question

我有一个并行度=256的运算符在128个任务管理器上运行。每次当我遇到检查点失败时，它都发生在该运算符的同一个子任务上，例如总是子任务129卡住并阻止检查点。我想通过检查子任务129运行的任务管理器的日志来了解这个子任务发生了什么。Flink中是否有方法将子任务id映射到相应的任务管理器？

Answer 1

我能够找到一个不平凡但可行的解决方案，以在运行时以编程方式获取所需的地图。

主要思想是Restendpoint/作业/：jobid/顶点/：vertexid以格式为特定顶点提供必要的信息

{
  "id": "804e...",
  "name": "Map -> Sink",
  ...
  "subtasks": [
    {
      "subtask": 0,
      "host": "ip-10-xx-yy-zz:36ddd"
    },
    ...
   ]
}

主要的困难是以编程方式获取web界面url。我可以这样得到它（可能有一个更优雅的解决方案）：

val env = FieldUtils
      .readField(getRuntimeContext.asInstanceOf[StreamingRuntimeContext], "taskEnvironment", true)
      .asInstanceOf[RuntimeEnvironment]

    try {
      println("trying to get cluster client...")
      val client = new RestClusterClient[String](env.getTaskManagerInfo.getConfiguration, "rest")
      return client.getWebInterfaceURL

    } catch {
      case e: Exception =>
        println("Failed to get cluster client : ")
        e.printStackTrace()
    }

给定Web界面url，我只需对其进行超文本传输协议调用并构建地图。

Answer 2

任务管理器。日志文件包含已部署任务的名称，包括其子任务索引。您只需在所有taskmanager中搜索TASK\u名称（129/256）。记录文件。