Apache Mesos 指南

最后修改:2024年1月8日

by baeldung

1. 概述

在同一个集群中部署多个应用是常见的做法。例如，我们通常会在一个集群中同时运行像 Apache Spark 或 Apache Flink 这样的分布式计算引擎，以及像 Apache Cassandra 这样的分布式数据库。

Apache Mesos 就是一个能有效协调这些应用之间资源共享的平台。

本文将先介绍在同一集群中资源分配的一些常见问题，然后说明 Apache Mesos 是如何优化资源利用率的。

2. 集群资源共享

多个应用共享一个集群时，通常有两种方式：

  • 静态划分集群资源，每个应用运行在指定分区
  • 为每个应用分配一组固定的机器

虽然这两种方式能让应用彼此独立运行，但资源利用率往往不高。

举个例子，某个应用可能只在短时间内运行，随后进入空闲状态。由于我们给它分配了固定的机器或分区，那么在空闲期间，这些资源就浪费了。

如果我们能在空闲期间将这些资源重新分配给其他应用，就能提升资源利用率。

Apache Mesos 正是通过动态资源分配来解决这个问题的。

3. Apache Mesos 简介

上面提到的两种资源共享方式中，应用只能感知到它运行的那部分机器资源。而 Mesos 提供了集群中所有资源的抽象视图，让应用可以基于整个集群的资源做调度决策。

Mesos 作为机器和应用之间的中间层，向应用提供集群中所有机器的可用资源，并定期更新资源状态，包括已完成任务释放的资源。这样，应用就能决定在哪台机器上执行哪个任务。

要理解 Mesos 的工作原理，我们先来看一下它的架构图：

Mesos 架构图

这是 Mesos 官方文档中的一张架构图（来源），图中 Hadoop 和 MPI 是两个共享集群的应用。

接下来我们逐一介绍图中各个组件。

3.1. Mesos Master

Mesos 的核心组件是 Master，它负责维护集群当前的资源状态，并在 Agent 和应用之间进行协调，传递资源和任务信息。

由于 Master 的故障会导致资源和任务状态丢失，因此通常会以高可用方式部署。如图所示，Mesos 会部署多个备用 Master，并通过 Zookeeper 来进行状态恢复。

3.2. Mesos Agent

Mesos 集群中的每台机器上都需要运行一个 Agent。Agent 会定期向 Master 汇报本地资源情况，并接收应用调度的任务。任务完成后或失败后，这个过程会重复。

后面我们会看到应用是如何在这些 Agent 上调度和执行任务的。

3.3. Mesos Frameworks

Mesos 允许应用实现一个抽象组件，与 Master 交互以获取集群中的可用资源，并基于这些资源做出调度决策。这些组件被称为 Frameworks。

一个 Framework 包含两个子组件：

  • Scheduler：基于所有 Agent 的可用资源，决定任务调度
  • Executor：运行在每个 Agent 上，执行调度任务

整个流程如下所示：

Mesos 流程图

首先，Agent 向 Master 汇报资源情况。Master 将这些资源提供给已注册的 Scheduler，这个过程称为 Resource Offer，我们会在下文详细讲解。

Scheduler 根据资源选择最合适的 Agent，通过 Master 执行任务。任务完成后，Executor 通知 Agent，Agent 再次上报资源给 Master。这个过程会在集群中所有 Framework 中重复进行。

Mesos 支持应用以多种语言实现自定义的 Scheduler 和 Executor。以 Java 为例，Scheduler 需要实现 Scheduler 接口：

public class HelloWorldScheduler implements Scheduler {
 
    @Override
    public void registered(SchedulerDriver schedulerDriver, Protos.FrameworkID frameworkID, 
      Protos.MasterInfo masterInfo) {
    }
 
    @Override
    public void reregistered(SchedulerDriver schedulerDriver, Protos.MasterInfo masterInfo) {
    }
 
    @Override
    public void resourceOffers(SchedulerDriver schedulerDriver, List<Offer> list) {
    }
 
    @Override
    public void offerRescinded(SchedulerDriver schedulerDriver, OfferID offerID) {
    }
 
    @Override
    public void statusUpdate(SchedulerDriver schedulerDriver, Protos.TaskStatus taskStatus) {
    }
 
    @Override
    public void frameworkMessage(SchedulerDriver schedulerDriver, Protos.ExecutorID executorID, 
      Protos.SlaveID slaveID, byte[] bytes) {
    }
 
    @Override
    public void disconnected(SchedulerDriver schedulerDriver) {
    }
 
    @Override
    public void slaveLost(SchedulerDriver schedulerDriver, Protos.SlaveID slaveID) {
    }
 
    @Override
    public void executorLost(SchedulerDriver schedulerDriver, Protos.ExecutorID executorID, 
      Protos.SlaveID slaveID, int i) {
    }
 
    @Override
    public void error(SchedulerDriver schedulerDriver, String s) {
    }
}

可以看到，这个接口主要是用于和 Master 通信的各种回调方法。

同理，Executor 需要实现 Executor 接口：

public class HelloWorldExecutor implements Executor {
    @Override
    public void registered(ExecutorDriver driver, Protos.ExecutorInfo executorInfo, 
      Protos.FrameworkInfo frameworkInfo, Protos.SlaveInfo slaveInfo) {
    }
  
    @Override
    public void reregistered(ExecutorDriver driver, Protos.SlaveInfo slaveInfo) {
    }
  
    @Override
    public void disconnected(ExecutorDriver driver) {
    }
  
    @Override
    public void launchTask(ExecutorDriver driver, Protos.TaskInfo task) {
    }
  
    @Override
    public void killTask(ExecutorDriver driver, Protos.TaskID taskId) {
    }
  
    @Override
    public void frameworkMessage(ExecutorDriver driver, byte[] data) {
    }
  
    @Override
    public void shutdown(ExecutorDriver driver) {
    }
}

稍后我们会看到 Scheduler 和 Executor 的实际运行效果。

4. 资源管理

4.1. 资源提供（Resource Offers）

如前所述，Agent 会定期将资源信息上报给 Master，Master 再将这些资源提供给集群中的 Framework。这个过程就是 Resource Offer。

每个 Resource Offer 包括两个部分：资源和属性。

  • 资源：用于上报 Agent 机器的硬件信息，如 CPU、内存、磁盘等。
  • 属性：用于提供额外的元数据，帮助 Framework 做调度决策。

Mesos 预定义了五种资源类型：

  • cpu
  • gpus
  • mem
  • disk
  • ports

资源值可以是以下三种类型之一：

  • Scalar：浮点数，例如 1.5G 内存
  • Range：数值范围，例如端口范围 [30000-34000]
  • Set：字符串集合

默认情况下，Mesos Agent 会自动检测这些资源，但我们也可以手动配置：

--resources='cpus:24;gpus:2;mem:24576;disk:409600;ports:[21000-24000,30000-34000];bugs(debug_role):{a,b,c}'

上面的例子中，除了标准资源，还配置了一个自定义资源 bugs，类型为 Set。

Agent 也可以发布键值对属性，例如：

--attributes='rack:abc;zone:west;os:centos5;level:10;keys:[1000-1500]'

属性值可以是 Scalar、Range 或文本类型。

4.2. 资源角色（Resource Roles）

现代操作系统通常支持多个用户，Mesos 也支持类似机制，称为 Role。每个 Role 可以看作是集群中的一个资源消费者。

Mesos Agent 可以根据不同策略将资源分配到不同 Role。Framework 也可以订阅这些 Role，实现对资源的细粒度控制。

例如，一个集群中有多个应用服务于不同用户，通过将资源划分到不同 Role，可以实现应用之间的隔离。

此外，Role 还能用于实现数据本地性。

举个例子，集群中有两个应用：producer 和 consumer。producer 写入数据到持久化卷，consumer 读取该卷。我们可以优化 consumer 的调度，让它在与 producer 相同的卷上运行：

  • 将持久化卷绑定到一个 Role
  • producer 和 consumer 的 Framework 都订阅该 Role
  • 这样 consumer 就可以在 producer 所在的卷上启动任务

4.3. 资源预留（Resource Reservation）

Mesos 通过 预留（Reservation） 将资源分配到不同的 Role。

预留分为两种：

  • 静态预留（Static Reservation）：在 Agent 启动时配置
 --resources="cpus:4;mem:2048;cpus(baeldung):8;mem(baeldung):4096"

上面例子中，为 Role baeldung 预留了 8 个 CPU 和 4096M 内存。

  • 动态预留（Dynamic Reservation）：允许在运行时通过 Framework 消息或 HTTP 接口重新分配资源。

没有指定 Role 的资源会被分配到默认 Role (*)，Master 会将这些资源提供给所有 Framework，无论它们是否订阅了该 Role。

4.4. 资源权重与配额（Resource Weights and Quotas）

Mesos 默认使用 加权主导资源公平策略（wDRF） 来公平分配资源。

但有时我们希望某些应用获得更多的资源。Mesos 允许 Framework 通过增加 Role 的权重来获得更多资源。

例如：

  • Role A 权重为 1
  • Role B 权重为 2
  • 那么 Role B 会获得两倍于 Role A 的资源

权重可以通过 HTTP 接口配置。

除了权重，Mesos 还支持 配额（Quota），即为某个 Role 保证最低资源数量。

5. 实现 Framework

如前所述，Mesos 支持用多种语言实现 Framework。在 Java 中，需要实现 Main 类、Scheduler 和 Executor。

5.1. Framework 主类

主类是 Framework 的入口点，负责：

  • 向 Master 注册
  • 提供 Executor 信息
  • 启动 Scheduler

首先添加 Maven 依赖：

<dependency>
    <groupId>org.apache.mesos</groupId>
    <artifactId>mesos</artifactId>
    <version>1.11.0</version>
</dependency>

接着实现 HelloWorldMain 类：

public static void main(String[] args) {
  
    String path = System.getProperty("user.dir")
      + "/target/libraries2-1.0.0-SNAPSHOT.jar";
  
    CommandInfo.URI uri = CommandInfo.URI.newBuilder().setValue(path).setExtract(false).build();
  
    String helloWorldCommand = "java -cp libraries2-1.0.0-SNAPSHOT.jar com.baeldung.mesos.executors.HelloWorldExecutor";
    CommandInfo commandInfoHelloWorld = CommandInfo.newBuilder()
      .setValue(helloWorldCommand)
      .addUris(uri)
      .build();
  
    ExecutorInfo executorHelloWorld = ExecutorInfo.newBuilder()
      .setExecutorId(Protos.ExecutorID.newBuilder()
      .setValue("HelloWorldExecutor"))
      .setCommand(commandInfoHelloWorld)
      .setName("Hello World (Java)")
      .setSource("java")
      .build();
}

这段代码配置了 Executor 的路径和启动命令。

继续初始化 Framework 并启动 Scheduler：

FrameworkInfo.Builder frameworkBuilder = FrameworkInfo.newBuilder()
  .setFailoverTimeout(120000)
  .setUser("")
  .setName("Hello World Framework (Java)");
 
frameworkBuilder.setPrincipal("test-framework-java");
 
MesosSchedulerDriver driver = new MesosSchedulerDriver(new HelloWorldScheduler(),
  frameworkBuilder.build(), args[0]);

最后启动 Driver：

int status = driver.run() == Protos.Status.DRIVER_STOPPED ? 0 : 1;
driver.stop();
System.exit(status);

其中 CommandInfo、ExecutorInfo 和 FrameworkInfo 都是 Protobuf 消息的 Java 表示。

5.2. 实现 Scheduler

从 Mesos 1.0 开始，可以通过 HTTP 接口与 Master 通信。

但如果是 Mesos 0.28 或更早版本，则需要实现 Scheduler 接口。

我们重点关注 resourceOffers 方法：

@Override
public void resourceOffers(SchedulerDriver schedulerDriver, List<Offer> list) {

    for (Offer offer : list) {
        List<TaskInfo> tasks = new ArrayList<TaskInfo>();
        Protos.TaskID taskId = Protos.TaskID.newBuilder()
          .setValue(Integer.toString(launchedTasks++)).build();

        System.out.println("Launching printHelloWorld " + taskId.getValue() + " Hello World Java");

        Protos.Resource.Builder cpus = Protos.Resource.newBuilder()
          .setName("cpus")
          .setType(Protos.Value.Type.SCALAR)
          .setScalar(Protos.Value.Scalar.newBuilder()
            .setValue(1));

        Protos.Resource.Builder mem = Protos.Resource.newBuilder()
          .setName("mem")
          .setType(Protos.Value.Type.SCALAR)
          .setScalar(Protos.Value.Scalar.newBuilder()
            .setValue(128));

这里我们为任务分配了 1 个 CPU 和 128M 内存。

接下来通过 SchedulerDriver 启动任务：

        TaskInfo printHelloWorld = TaskInfo.newBuilder()
          .setName("printHelloWorld " + taskId.getValue())
          .setTaskId(taskId)
          .setSlaveId(offer.getSlaveId())
          .addResources(cpus)
          .addResources(mem)
          .setExecutor(ExecutorInfo.newBuilder(helloWorldExecutor))
          .build();

        List<OfferID> offerIDS = new ArrayList<>();
        offerIDS.add(offer.getId());

        tasks.add(printHelloWorld);

        schedulerDriver.launchTasks(offerIDS, tasks);
    }
}

如果 Scheduler 无法在某个 Agent 上启动任务，应立即拒绝该资源 Offer：

schedulerDriver.declineOffer(offer.getId());

5.3. 实现 Executor

Executor 负责在 Agent 上执行任务。

我们之前配置了启动命令：

java -cp libraries2-1.0.0-SNAPSHOT.jar com.baeldung.mesos.executors.HelloWorldExecutor

其中 HelloWorldExecutor 是主类：

public class HelloWorldExecutor implements Executor {
    public static void main(String[] args) {
        MesosExecutorDriver driver = new MesosExecutorDriver(new HelloWorldExecutor());
        System.exit(driver.run() == Protos.Status.DRIVER_STOPPED ? 0 : 1);
    }
}

最后实现任务执行逻辑：

public void launchTask(ExecutorDriver driver, TaskInfo task) {
 
    Protos.TaskStatus status = Protos.TaskStatus.newBuilder()
      .setTaskId(task.getTaskId())
      .setState(Protos.TaskState.TASK_RUNNING)
      .build();
    driver.sendStatusUpdate(status);
 
    System.out.println("Execute Task!!!");
 
    status = Protos.TaskStatus.newBuilder()
      .setTaskId(task.getTaskId())
      .setState(Protos.TaskState.TASK_FINISHED)
      .build();
    driver.sendStatusUpdate(status);
}

Executor 也可以发送消息回 Scheduler：

String myStatus = "Hello Framework";
driver.sendFrameworkMessage(myStatus.getBytes());

6. 总结

本文简要介绍了集群中多个应用之间的资源共享问题，并说明了 Apache Mesos 如何通过抽象资源视图实现资源的最大化利用。

我们还讨论了基于不同公平策略和角色的动态资源分配机制，以及 Mesos 是如何让应用基于资源 Offer 做出调度决策的。

最后，我们用 Java 实现了一个简单的 Mesos Framework，包括 Scheduler 和 Executor。

完整示例代码可在 GitHub 上找到。

« 上一篇： Java 中字符串反转的实现方法
» 下一篇： Spring @Qualifier 注解详解