org.apache.flink.yarn.YarnResourceManager类的使用及代码示例

x33g5p2x  于2022-02-05 转载在 其他  
字(10.9k)|赞(0)|评价(0)|浏览(234)

本文整理了Java中org.apache.flink.yarn.YarnResourceManager类的一些代码示例，展示了YarnResourceManager类的具体用法。这些代码示例主要来源于Github/Stackoverflow/Maven等平台，是从一些精选项目中提取出来的代码，具有较强的参考意义，能在一定程度帮忙到你。YarnResourceManager类的具体详情如下：
包路径：org.apache.flink.yarn.YarnResourceManager
类名称：YarnResourceManager

YarnResourceManager介绍

[英]The yarn implementation of the resource manager. Used when the system is started via the resource framework YARN.
[中]资源管理器的实现。通过资源框架启动系统时使用。

代码示例

代码示例来源：origin: apache/flink

rpcService.getScheduledExecutor());
return new YarnResourceManager(
  rpcService,
  ResourceManager.RESOURCE_MANAGER_NAME,

代码示例来源：origin: apache/flink

@Override
public void onContainersCompleted(final List<ContainerStatus> statuses) {
  runAsync(() -> {
      log.debug("YARN ResourceManager reported the following containers completed: {}.", statuses);
      for (final ContainerStatus containerStatus : statuses) {
        final ResourceID resourceId = new ResourceID(containerStatus.getContainerId().toString());
        final YarnWorkerNode yarnWorkerNode = workerNodeMap.remove(resourceId);
        if (yarnWorkerNode != null) {
          // Container completed unexpectedly ~> start a new one
          requestYarnContainerIfRequired();
        }
        // Eagerly close the connection with task manager.
        closeTaskManagerConnection(resourceId, new Exception(containerStatus.getDiagnostics()));
      }
    }
  );
}

代码示例来源：origin: apache/flink

@Override
protected void initialize() throws ResourceManagerException {
  try {
    resourceManagerClient = createAndStartResourceManagerClient(
      yarnConfig,
      yarnHeartbeatIntervalMillis,
      webInterfaceUrl);
  } catch (Exception e) {
    throw new ResourceManagerException("Could not start resource manager client.", e);
  }
  nodeManagerClient = createAndStartNodeManagerClient(yarnConfig);
}

代码示例来源：origin: apache/flink

/**
 * Request new container if pending containers cannot satisfies pending slot requests.
 */
private void requestYarnContainerIfRequired() {
  int requiredTaskManagerSlots = getNumberRequiredTaskManagerSlots();
  int pendingTaskManagerSlots = numPendingContainerRequests * numberOfTaskSlots;
  if (requiredTaskManagerSlots > pendingTaskManagerSlots) {
    requestYarnContainer();
  }
}

代码示例来源：origin: apache/flink

@Override
public void onContainersAllocated(List<Container> containers) {
  runAsync(() -> {
    final Collection<AMRMClient.ContainerRequest> pendingRequests = getPendingRequests();
    final Iterator<AMRMClient.ContainerRequest> pendingRequestsIterator = pendingRequests.iterator();
        removeContainerRequest(pendingRequestsIterator.next());
          ContainerLaunchContext taskExecutorLaunchContext = createTaskExecutorLaunchContext(
            container.getResource(),
            containerIdStr,
          resourceManagerClient.releaseAssignedContainer(container.getId());
          requestYarnContainerIfRequired();

代码示例来源：origin: apache/flink

protected AMRMClientAsync<AMRMClient.ContainerRequest> createAndStartResourceManagerClient(
    YarnConfiguration yarnConfiguration,
    int yarnHeartbeatIntervalMillis,
    @Nullable String webInterfaceUrl) throws Exception {
  AMRMClientAsync<AMRMClient.ContainerRequest> resourceManagerClient = AMRMClientAsync.createAMRMClientAsync(
    yarnHeartbeatIntervalMillis,
    this);
  resourceManagerClient.init(yarnConfiguration);
  resourceManagerClient.start();
  //TODO: change akka address to tcp host and port, the getAddress() interface should return a standard tcp address
  Tuple2<String, Integer> hostPort = parseHostPort(getAddress());
  final int restPort;
  if (webInterfaceUrl != null) {
    final int lastColon = webInterfaceUrl.lastIndexOf(':');
    if (lastColon == -1) {
      restPort = -1;
    } else {
      restPort = Integer.valueOf(webInterfaceUrl.substring(lastColon + 1));
    }
  } else {
    restPort = -1;
  }
  final RegisterApplicationMasterResponse registerApplicationMasterResponse =
    resourceManagerClient.registerApplicationMaster(hostPort.f0, restPort, webInterfaceUrl);
  getContainersFromPreviousAttempts(registerApplicationMasterResponse);
  return resourceManagerClient;
}

代码示例来源：origin: org.apache.flink/flink-yarn

@Override
public void onContainersAllocated(List<Container> containers) {
  runAsync(() -> {
    for (Container container : containers) {
      log.info(
          ContainerLaunchContext taskExecutorLaunchContext = createTaskExecutorLaunchContext(
            container.getResource(),
            containerIdStr,
          resourceManagerClient.releaseAssignedContainer(container.getId());
          requestYarnContainerIfRequired(container.getResource(), container.getPriority());

代码示例来源：origin: apache/flink

@Override
protected void internalDeregisterApplication(
    ApplicationStatus finalStatus,
    @Nullable String diagnostics) {
  // first, de-register from YARN
  FinalApplicationStatus yarnStatus = getYarnStatus(finalStatus);
  log.info("Unregister application from the YARN Resource Manager with final status {}.", yarnStatus);
  try {
    resourceManagerClient.unregisterApplicationMaster(yarnStatus, diagnostics, "");
  } catch (Throwable t) {
    log.error("Could not unregister the application master.", t);
  }
  Utils.deleteApplicationFiles(env);
}

代码示例来源：origin: apache/flink

this.resource = Resource.newInstance(defaultTaskManagerMemoryMB, defaultCpus);
this.slotsPerWorker = createSlotsPerWorker(numberOfTaskSlots);

代码示例来源：origin: org.apache.flink/flink-yarn_2.11

@Override
public void onContainersAllocated(List<Container> containers) {
  runAsync(() -> {
    final Collection<AMRMClient.ContainerRequest> pendingRequests = getPendingRequests();
    final Iterator<AMRMClient.ContainerRequest> pendingRequestsIterator = pendingRequests.iterator();
        removeContainerRequest(pendingRequestsIterator.next());
          ContainerLaunchContext taskExecutorLaunchContext = createTaskExecutorLaunchContext(
            container.getResource(),
            containerIdStr,
          resourceManagerClient.releaseAssignedContainer(container.getId());
          requestYarnContainerIfRequired();

代码示例来源：origin: org.apache.flink/flink-yarn

protected AMRMClientAsync<AMRMClient.ContainerRequest> createAndStartResourceManagerClient(
    YarnConfiguration yarnConfiguration,
    int yarnHeartbeatIntervalMillis,
    @Nullable String webInterfaceUrl) throws Exception {
  AMRMClientAsync<AMRMClient.ContainerRequest> resourceManagerClient = AMRMClientAsync.createAMRMClientAsync(
    yarnHeartbeatIntervalMillis,
    this);
  resourceManagerClient.init(yarnConfiguration);
  resourceManagerClient.start();
  //TODO: change akka address to tcp host and port, the getAddress() interface should return a standard tcp address
  Tuple2<String, Integer> hostPort = parseHostPort(getAddress());
  final int restPort;
  if (webInterfaceUrl != null) {
    final int lastColon = webInterfaceUrl.lastIndexOf(':');
    if (lastColon == -1) {
      restPort = -1;
    } else {
      restPort = Integer.valueOf(webInterfaceUrl.substring(lastColon + 1));
    }
  } else {
    restPort = -1;
  }
  final RegisterApplicationMasterResponse registerApplicationMasterResponse =
    resourceManagerClient.registerApplicationMaster(hostPort.f0, restPort, webInterfaceUrl);
  getContainersFromPreviousAttempts(registerApplicationMasterResponse);
  return resourceManagerClient;
}

代码示例来源：origin: org.apache.flink/flink-yarn_2.11

/**
 * Request new container if pending containers cannot satisfies pending slot requests.
 */
private void requestYarnContainerIfRequired() {
  int requiredTaskManagerSlots = getNumberRequiredTaskManagerSlots();
  int pendingTaskManagerSlots = numPendingContainerRequests * numberOfTaskSlots;
  if (requiredTaskManagerSlots > pendingTaskManagerSlots) {
    requestYarnContainer();
  }
}

代码示例来源：origin: org.apache.flink/flink-yarn_2.11

@Override
protected void internalDeregisterApplication(
    ApplicationStatus finalStatus,
    @Nullable String diagnostics) {
  // first, de-register from YARN
  FinalApplicationStatus yarnStatus = getYarnStatus(finalStatus);
  log.info("Unregister application from the YARN Resource Manager with final status {}.", yarnStatus);
  try {
    resourceManagerClient.unregisterApplicationMaster(yarnStatus, diagnostics, "");
  } catch (Throwable t) {
    log.error("Could not unregister the application master.", t);
  }
  Utils.deleteApplicationFiles(env);
}

代码示例来源：origin: org.apache.flink/flink-yarn

this.defaultCpus = flinkConfig.getInteger(YarnConfigOptions.VCORES, numberOfTaskSlots);
this.slotsPerWorker = createSlotsPerWorker(numberOfTaskSlots);

代码示例来源：origin: org.apache.flink/flink-yarn_2.11

@Override
public void onContainersCompleted(final List<ContainerStatus> statuses) {
  runAsync(() -> {
      log.debug("YARN ResourceManager reported the following containers completed: {}.", statuses);
      for (final ContainerStatus containerStatus : statuses) {
        final ResourceID resourceId = new ResourceID(containerStatus.getContainerId().toString());
        final YarnWorkerNode yarnWorkerNode = workerNodeMap.remove(resourceId);
        if (yarnWorkerNode != null) {
          // Container completed unexpectedly ~> start a new one
          requestYarnContainerIfRequired();
        }
        // Eagerly close the connection with task manager.
        closeTaskManagerConnection(resourceId, new Exception(containerStatus.getDiagnostics()));
      }
    }
  );
}

代码示例来源：origin: org.apache.flink/flink-yarn_2.11

protected AMRMClientAsync<AMRMClient.ContainerRequest> createAndStartResourceManagerClient(
    YarnConfiguration yarnConfiguration,
    int yarnHeartbeatIntervalMillis,
    @Nullable String webInterfaceUrl) throws Exception {
  AMRMClientAsync<AMRMClient.ContainerRequest> resourceManagerClient = AMRMClientAsync.createAMRMClientAsync(
    yarnHeartbeatIntervalMillis,
    this);
  resourceManagerClient.init(yarnConfiguration);
  resourceManagerClient.start();
  //TODO: change akka address to tcp host and port, the getAddress() interface should return a standard tcp address
  Tuple2<String, Integer> hostPort = parseHostPort(getAddress());
  final int restPort;
  if (webInterfaceUrl != null) {
    final int lastColon = webInterfaceUrl.lastIndexOf(':');
    if (lastColon == -1) {
      restPort = -1;
    } else {
      restPort = Integer.valueOf(webInterfaceUrl.substring(lastColon + 1));
    }
  } else {
    restPort = -1;
  }
  final RegisterApplicationMasterResponse registerApplicationMasterResponse =
    resourceManagerClient.registerApplicationMaster(hostPort.f0, restPort, webInterfaceUrl);
  getContainersFromPreviousAttempts(registerApplicationMasterResponse);
  return resourceManagerClient;
}

代码示例来源：origin: org.apache.flink/flink-yarn

/**
 * Request new container if pending containers cannot satisfies pending slot requests.
 */
private void requestYarnContainerIfRequired(Resource resource, Priority priority) {
  int requiredTaskManagerSlots = getNumberRequiredTaskManagerSlots();
  int pendingTaskManagerSlots = numPendingContainerRequests * numberOfTaskSlots;
  if (requiredTaskManagerSlots > pendingTaskManagerSlots) {
    requestYarnContainer(resource, priority);
  }
}

代码示例来源：origin: org.apache.flink/flink-yarn_2.11

@Override
protected void initialize() throws ResourceManagerException {
  try {
    resourceManagerClient = createAndStartResourceManagerClient(
      yarnConfig,
      yarnHeartbeatIntervalMillis,
      webInterfaceUrl);
  } catch (Exception e) {
    throw new ResourceManagerException("Could not start resource manager client.", e);
  }
  nodeManagerClient = createAndStartNodeManagerClient(yarnConfig);
}

代码示例来源：origin: org.apache.flink/flink-yarn

@Override
protected void internalDeregisterApplication(
    ApplicationStatus finalStatus,
    @Nullable String diagnostics) {
  // first, de-register from YARN
  FinalApplicationStatus yarnStatus = getYarnStatus(finalStatus);
  log.info("Unregister application from the YARN Resource Manager with final status {}.", yarnStatus);
  try {
    resourceManagerClient.unregisterApplicationMaster(yarnStatus, diagnostics, "");
  } catch (Throwable t) {
    log.error("Could not unregister the application master.", t);
  }
  Utils.deleteApplicationFiles(env);
}

代码示例来源：origin: org.apache.flink/flink-yarn

rpcService.getScheduledExecutor());
return new YarnResourceManager(
  rpcService,
  ResourceManager.RESOURCE_MANAGER_NAME,

内容来源于网络，如有侵权，请联系作者删除！

YarnResourceManager

相关文章

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新文章

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com