深入浅出 Yarn Application Master 编写

后端

2024-01-16 13:43:14

在上一篇文章中，我们探讨了 Yarn Application 中 Client 部分的编写方法。本篇将继续介绍 ApplicationMaster 部分的编写。

Application Master 编写方法

Application Master 是 Yarn 中负责协调和管理 Application 的组件。它负责以下任务：

协商资源
启动和监控容器
跟踪应用程序进度
处理故障

要在 Yarn 中编写 Application Master，需要遵循以下步骤：

1. 创建一个 Application Master 类

Application Master 类必须扩展 Yarn 提供的 ApplicationMaster 类。该类必须实现以下方法：

run 方法：应用程序主体的执行逻辑
handle 方法：处理来自 Yarn 的消息

2. 实现 run 方法

run 方法是 Application Master 的入口点。该方法负责执行应用程序的主体逻辑，包括：

协商资源
启动和监控容器
跟踪应用程序进度
处理故障

3. 实现 handle 方法

handle 方法用于处理来自 Yarn 的消息。该方法负责处理来自 ResourceManager 和 NodeManager 的消息。

4. 提交应用程序

要提交应用程序，需要使用 Yarn 提供的 Client 类。Client 类提供了一系列方法来提交和管理应用程序。

示例

以下是一个简单的 Application Master 类的示例：

import org.apache.hadoop.yarn.api.records.*;
import org.apache.hadoop.yarn.client.api.AMRMClient;
import org.apache.hadoop.yarn.client.api.AMRMClient.ContainerRequest;
import org.apache.hadoop.yarn.client.api.YarnClient;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.exceptions.YarnException;
import org.apache.hadoop.yarn.util.Records;

import java.io.IOException;
import java.util.List;

public class MyApplicationMaster implements ApplicationMaster {

    private YarnClient yarnClient;
    private AMRMClient<ContainerRequest> resourceManager;
    private String applicationId;

    public MyApplicationMaster() {
        this.yarnClient = YarnClient.createYarnClient();
        this.yarnClient.init(new YarnConfiguration());
        this.resourceManager = yarnClient.createAMRMClient();
        this.resourceManager.init(new YarnConfiguration());
        this.applicationId = yarnClient.getApplicationId();
    }

    @Override
    public void run() {
        // 协商资源
        Resource resource = Records.newRecord(Resource.class);
        resource.setMemory(1024);
        resource.setVirtualCores(1);
        ContainerRequest containerRequest = Records.newRecord(ContainerRequest.class);
        containerRequest.setResource(resource);
        containerRequest.setPriority(Priority.newInstance(0));

        List<ContainerRequest> containerRequests = Collections.singletonList(containerRequest);
        try {
            resourceManager.addContainerRequests(containerRequests);
        } catch (YarnException | IOException e) {
            e.printStackTrace();
        }

        // 启动和监控容器
        while (true) {
            try {
                // 处理来自 Yarn 的消息
                YarnApplicationAttempt attempt = resourceManager.getApplicationAttempt(applicationId);
                ApplicationMasterServiceContext context = attempt.getServiceContext();
                resourceManager.handle(context.getNMToken(), null);

                // 检查应用程序状态
                ApplicationReport report = yarnClient.getApplicationReport(applicationId);
                if (report.getFinalApplicationStatus() != null) {
                    break;
                }
            } catch (YarnException | IOException e) {
                e.printStackTrace();
            }
        }

        // 处理故障

        // 跟踪应用程序进度
    }

    @Override
    public void handle(ContainerStatusRequest containerStatusRequest) {
        // 处理来自 NodeManager 的消息
    }

    public static void main(String[] args) {
        MyApplicationMaster appMaster = new MyApplicationMaster();
        appMaster.run();
    }
}