Doris On Yarn

本文讲述基于 Yarn 动态的部署 Doris BE 服务，以实现 Doris 和 Hadoop 集群资源利用率最大化。

背景

目前 Doris 在公司内共部署了 3 个集群，具体信息如下：

• 测试集群

  提供给 Doris 控制台做建表脚本测试，稳定性保障不高。

• 报表集群

  数据分析人员使用，通过将数仓部分 Hive 表直接同步到该集群，并且通过一些 BI 工具（帆软 、观远 等）做报表分析，数据规模不大，但是稳定性需要严格保障。

• 标签集群

  目前是给离线标签集群替换 es 来作为高效查询使用，因为是线上业务，稳定性也需要严格保障。

自助查询规划通过 Doris 加速查询性能，标签集群最开始是规划给自助查询使用，但是因为数据导入成本太高放弃使用，但还是需要一种方式能够对自助查询提供加速。

Doris

Doris 1.2.0 后新增两个特性

• 多源数据目录（Multi-Catalog）

  旨在能够更方便对接外部数据目录，以增强 Doris 的数据湖分析和联邦数据查询能力。构成 Catalog -> Database -> Table 的三层元数据层级，Catalog 可以直接对应到外部数据目录，很方便拿到所有 Database 映射。

• 计算节点（computation）

  计算节点专门处理数据湖这类远程的联邦查询. 原来的 BE 节点类型称为混合节点, 这类节点既能做 SQL 查询, 又有 Tablet 数据存储管理. 而计算节点只能做 SQL 查询, 它不会保存任何数据.

混合节点又可以查询，又可以存储，为什么还需要计算节点？

• 资源隔离差, 两个负载对集群的响应要求不一, 混合部署会有相互的影响.
• 集群扩容时, 数据湖查询只需要扩容计算资源, 而目前只能存储计算一起扩容, 导致磁盘使用率变低.
• 扩容效率差, 扩容后会启动 Tablet 数据的迁移, 整体过程比较漫长. 而数据湖查询有着明显的高峰低谷, 需要小时级弹性能力.

同时由于没有多余的资源部署计算节点，计算节点是无状态的，考虑资源合理使用的问题，我们需要探索是否可以基于 Hadoop Yarn 动态的部署 Doris BE 服务，以实现 Doris 和 Hadoop 集群资源的最大化复用, 这里使用 Skein 部署 Doris BE 服务。

Skein

Skein 是一个用于在 Apache Hadoop 集群上运行和管理分布式应用程序的开源框架

架构

20230523030041

• Application Master（应用程序主节点）：
  Skein 的核心组件是 Application Master，它是一个运行在 YARN 上的 Java 进程，负责解释和执行应用程序的规范（Specification）。Application Master 管理应用程序的整个生命周期，包括资源的申请和分配、容器的启动和监控、任务的调度和执行等。它与 YARN ResourceManager 进行通信，并协调各个任务的执行。

• Specification（规范）：
  Skein 使用规范来描述和定义应用程序的各个组件、资源需求、启动命令等。规范以 JSON 格式表示，并包含了应用程序的配置信息、依赖项、服务定义等。Application Master 根据规范来启动和管理应用程序。

• Services（服务）：
  Skein 的规范可以定义多个服务，每个服务代表一个应用程序组件或服务。服务可以是一个用户定义的进程、容器、Web 服务等。Application Master 负责启动、监控和管理这些服务，并提供与它们的交互接口。

• Container（容器）：
  在 Skein 中，容器是应用程序的基本执行单元。Application Master 通过向 YARN 请求资源来获取容器，并在这些容器中运行应用程序的服务和任务。每个容器都有自己的资源限制和环境设置，可以独立地执行任务。

总体而言，Skein 的架构包括一个 Application Master 负责应用程序的管理，规范定义应用程序的组件和配置，服务代表应用程序的各个组件，容器是执行任务的基本单位。通过这种架构，Skein 实现了在 YARN 上运行和管理分布式应用程序的能力。

skein 配置文件

name: doris
queue: doris
node_label: doris
master:
  log_level: debug
services:
  doris.be:
    instances: 128
    resources:
      memory: 64 GiB
      vcores: 8
    # 服务允许的最大重启次数，-1 以始终重新启动
    max_restarts: -1
    # 运行服务所需的文件
    files:
      doris_be: hdfs:///user/bolt/doris/doris-be-1.2.4.tar.gz
      doris_sh: hdfs:///user/bolt/doris/doris-be.sh
      be_conf: hdfs:///user/bolt/doris/be.conf
    # 运行服务的bash脚本
    script: |
      sh doris_sh $(readlink -f doris_be) $(readlink -f be_conf)
      exit $?
SHELL

从上述配置问价可看出，这里会部署 128 台 8C64G 的 container 到 Yarn 中的 Doris 队列 Doris 标签节点上，启动脚本为 doris_sh。

doris-be.sh

#!/bin/bash

DORIS_BE_PATH=$1
CONF_DIR=$2

DORIS_VERSION=1.2.4
LOG_PATH="/tmp/doris/log"

# 输出脚本日志到 /doris-on-yarn.log
function echo_log() {
  echo "$(date "+%Y-%m-%d %H:%M:%S"): $@" | tee -a "${LOG_PATH}/doris-on-yarn.log"
}

mkdir -p "${LOG_PATH}"

echo_log "${DORIS_BE_PATH}"
echo_log "${CONF_DIR}"

# set conf
/bin/cp -f ${CONF_DIR} "${DORIS_BE_PATH}/apache-doris-be-${DORIS_VERSION}-bin-x86_64/conf/be.conf"
sed -i 's#export LOG_DIR="\${DORIS_HOME}/log"#export LOG_DIR="/tmp/doris/log"#g' "${DORIS_BE_PATH}/apache-doris-be-${DORIS_VERSION}-bin-x86_64/bin/start_be.sh"

# Start the BE
echo_log "-------- Starting Doris BE $(hostname -I) --------"
"${DORIS_BE_PATH}/apache-doris-be-${DORIS_VERSION}-bin-x86_64/bin/start_be.sh"
exit $?
SHELL

1. 复制自定义 be.conf 到 doris conf 目录
2. container 会销毁，不便于查看日志，因此把日志输出到机器的 /tmp/doris/log 下
3. 启动 Doris-be

为什么要 exit $?

配置了 max_restarts: -1 container 失败一直重启，需要把失败状态码传递到 Application Master，才会重新分配资源重新启动。

相关命令

# 启动命令
skein submit doris-be.yaml
# 查看 application
skein application ls
# 停止
skein shutdown <APP_ID>
SHELL

Yarn 资源分配策略

在实践中发现，由于并不是 128 台 Be 都启动了，因此需要去了解 Yarn 的资源分配策略进行优化

在 YARN 中，Application 的 Container 异常崩溃后不会自动恢复。当一个 Container 发生异常崩溃时，YARN 会将该 Container 标记为失败，并通知 ResourceManager。ResourceManager 会记录该 Container 的失败状态，并可以采取相应的操作，例如重新分配资源或记录日志。

以下是一般情况下 YARN 重新分配资源的策略：

1. 资源重新分配：ResourceManager 会尝试重新为失败的 Container 分配资源，以满足 Application 的需求。资源的重新分配可能会在同一节点上或其他可用节点上进行。
2. 容器优先级：重新分配资源时，ResourceManager 会考虑容器的优先级。较高优先级的容器可能会被优先重新分配资源，以确保重要的任务能够快速恢复。
3. 容器位置约束：根据配置和调度策略，ResourceManager 可能会尽量将重新分配的容器分配到特定的节点或节点标签上，以满足应用程序的需求。
4. 容器重试限制：ResourceManager 通常会限制容器的重试次数。如果一个容器连续失败超过预设的重试次数限制，ResourceManager 可能会认为该容器无法成功运行，并不再重新分配资源给它。（目前 doris 失败重试过一段时间不在继续重试原因）

资源调度器

FIFO Scheduler(先进先出调度器)

20230523022449

将所有的 Application 按照提交时候的顺序来执行，只有当上一个 Job 执行完成之后后面的 Job 才会按照队列的顺序依次被执行。FIFO 调度器以集群资源独占的方式来运行作业，后面的 Job 会被阻塞。简单但不实用。

Fair Scheduler(公平调度器)

20230523022322

所有的 Job 平均的获取资源。默认情况下，Fair 调度器只是对内存资源做公平的调度和分配。多个队列会平分资源，队列内的 Job 也会平分资源。

Capacity Scheduler(容量调度器)（我们使用的）

image.png

相关参数：

• capacity：队列的资源容量（百分比）。 当系统非常繁忙时，应保证每个队列的容量得到满足，而如果每个队列应用程序较少，可将剩余资源共享给其他队列。

• maximum-capacity：队列的资源使用上限（百分比）。由于存在资源共享，因此一个队列使用的资源量可能超过其容量，而最多使用资源量可通过该参数限制。（这也是前文提到的关于有任务运行的队列可以占用的资源的最大百分比）

• user-limit-factor：每个用户最多可使用的队列资源量，默认 1（100%）。

假如整个集群有 10 台 100G 机器，其中有 5 台 label 为 doris，剩余 5 台 label 为 other，A 队列配置 label 为 doris，B 队列配置 label 为 other，c 队列配置 label 为 doris 和 other，A 队列 Capacity 为 10%,B 队列 Capacity 为 10%,C 队列 Capacity 为 10%，请问 A 队列实际能使用的容量是多少？

A: 5 * 100G * 10% = 50GB
B: 5 * 100G * 10% = 50GB
C: (5+5) * 100G * 10% = 100GB
SHELL

我们实际 Doris 队列原本 Capacity 为 10%，不足以满足 128 个 Be 内存需求，有上面例子可以计算出我们实际需要的 Capacity 为 25%（具体数据不列举），同时运行的 Be 平均 70 个增加到 100 个，理想情况下能够同时运行 128 个 Be（只发现过一次）。

为什么很难到达 128 个呢，由于端口冲突一台 node 只能起一个 Doris container，ResourceManager 分配是按容量分配资源，而不是按每个 Node 分配一个 container，以及 ResourceManager 容器重试次数限制，因此大部分情况下并不能 128 个都启动。

总结

本文简单介绍了如何基于 Yarn 动态的部署 Doris BE 服务，同时了解到 Yarn 资源分配策略。这对我们理解大数据生态带来一些启发。

参考

1. https://jcristharif.com/skein/quickstart.html
2. https://doris.apache.org/zh-CN/
3. https://zhuanlan.zhihu.com/p/27953843