本文是 yarn 学习笔记,主要参考 《Hadoop技术内幕:深入解析YARN架构设计与实现原理》,对比 yarn 和 kubernetes 的实现差异。
Yarn 两个重要的组件 RM 和 NM:
注意 AM 一般指的是 ApplicationMaster 不是 ApplicationManager
ApplicationMaster(AM),用户提交的每个应用程序都需要包含一个AM, 作用为:
带有有限状态机的事件处理器
,其处理结果也以事 件的形式输出给中央异步调度器代码位置
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/api
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/server/api
hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-common/src/main/java/org/apache/hadoop/yarn/server/api
graph LR
JobClient-->|ApplicationClientProtocol|RM
Admin-->|ResourceManagerAdministrationProtocol|RM
AM-->|ApplicationMasterProtocol|RM
AM-->|ContainerManagementProtocol|NM
NM-->|ResourceTracker|RM
FROM -> TO | 主要作用 | 重要接口 |
---|---|---|
JobClient -> RM (ApplicationClientProtocol) | JobClient 通过该RPC提交应用程序、查询应用程序状态等 |
|
Admin -> RM (ResourceManagerAdministrationProtocol) | Admin 通过该RPC更新系统配置文件,比如节点黑白名单、用户队列权限等 |
|
AM -> RM (ApplicationMasterProtocol) | AM 通过该 RPC 向RM注册和注销自己,并为各个任务申请资源 |
|
AM -> NM (ContainerManagementProtocol) | AM 通过该 RPC 要求 NM 启动或停 止Container,获取各个 container 的状态等信息 |
|
NM -> RM (ResourceTracker) | NM 通过该 RPC 协议向 RM 注册,并定时发送心跳汇报当前节点的资源使用情况和 Container 运行情况 |
|
基本运行流程
sequenceDiagram
Client->>RM: 1. 提交应用,其中包括AM、启动AM的命令、用户程序等
RM->>AM: 2. 分配第一个Container,与对应的NM通信,要求它在这个Container中启动应用AM
AM->>RM: 3. 向RM注册,用户可以通过RM查看应用状态。AM为各个任务申请资源,控运行状态到运行结束
loop
AM->>RM: 4. 采用轮询的方式通过RPC协议向RM申请和领取资源
AM->>NM: 5. 获得资源后,便与对应的NM通信,要求它启动任务
NM->>Container: 6. NM为任务设置好运行环境(包括环境变量/JAR包/二进制程序等)后,将任务启动命令写到一个脚本中,并通过其启动任务
Container->>AM: 7. 通过RPC协议向AM汇报自己的状态/进度,以让AM掌握状态,从而可以在任务失败时重启任务
end
AM->>RM: 8.申请注销并关闭自己
思考: AM 可不可以省略,集成到 RM 成为一个线程 (插件),让整个架构变得更简单清晰?
概述见上,RM 中的 Service 分为 "Always On" services 和 "Active" services,表示 HA 模式 Leader 的功能;
从多个模块角度看:
ApplicationClientProtocol
, 提供一个 rpc server (多种实现 RpcEngine,主要:ProtobufRpcEngine2) 为普通用户提供服务,它处理来自客户端的各种 RPC,比如:终止/提交应用/获取应用状态等,内部会调用如 RMContext
(最重要,内部有 Dispatcher/HAServiceState/RMStateStore 等等); RMAppManager
; RMContext.Dispatcher
(异步请求通过事件);ResourceScheduler
; YarnScheduler
完成请求ResourceTracker
协议: 处理来自NodeManager的请求,主要包括:ApplicationMasterProtocol
(注意 AM 和 RMApp 是一一对应的): 处理来自AM的请求,包括:sequenceDiagram
ApllicationMasterLauncher->>NodeManager: 1. StartContainer启动app的ApplicationMaster
ApllicationMasterLauncher->>AMLivelinessMonitor: 2.通过事件将 AM 注册到AMLivelinessMonitor, 启动心跳监控
ApplicationMaster->>ApplicationMasterService: 3. 注册自己
loop
ApplicationMaster->>ApplicationMasterService: 4. allocate 心跳
ApplicationMasterService->>AMLivelinessMonitor: 5. update exprire time
end
ApplicationMaster->>ApplicationMasterService: 6. 注销自己
ApplicationMasterService->>AMLivelinessMonitor: 7. remove AM 心跳监控
序号 | 组件名称 | 服务/事件处理器 | 处理的事件类型 | 输出事件类型 |
---|---|---|---|---|
1 | ClientRMService | 服务 | – | RMAppAttemptEvent/RMAppEvent/RMNodeEvent |
2 | NMLivelinessMonitor | 服务 | – | RMNodeEvent |
3 | ResourceTrackerService | 服务 | – | RMNodeEvent/RMAppAttemptEvent |
4 | AMLivelinessMonitor | 服务 | – | RMAppAttemptEvent |
5 | ContainerAllocationExpirer | 服务 | – | SchedulerEvent |
6 | ApplicationMasterLauncher | 事件处理器 | AMLauncherEvent | – |
7 | RMAppManager | 事件处理器 | RMAppManagerEvent | RMAppEvent |
8 | NodesListManager | 事件处理器 | NodesListManagerEvent | RMNodeEvent/RMAppEvent |
9 | RMApp(ApplicationEventDispatcher)(RMAppImpl) | 事件处理器 | RMAppEvent | RMAppAttemptEvent/RMNodeEvent/SchedulerEvent/RMAppManagerEvent |
10 | RMAppAttempt(ApplicationAttemptEventDispatcher)(RMAppAttemptImpl) | 事件处理器 | RMAppAttemptEvent | SchedulerEvent/RMAppAttemptEvent/RMAppEvent/AMLauncherEvent/RMNodeEvent |
11 | RMNode(NodeEventDispatcher)(RMNodeImpl) | 事件处理器 | RMNodeEvent | RMAppEvent/SchedulerEvent/NodesListManagerEvent/RMNodeEvent |
12 | ResourceScheduler(EventDispatcher)(FairScheduler) | 事件处理器 | SchedulerEvent | RMAppEvent/RMAppAttemptEvent |
13 | RMContainer (RMContainerImpl) | 事件处理器(非异步) | RMContainerEvent | RMAppEvent/RMAppAttemptEvent/RMNodeEvent |
以 MapReduce 任务为例,实现了自己 的 Client(JobClient) 和 MRAppMaster(AM)
sequenceDiagram
Client->>ResourceManager: ApplicationClientProtocol:forceKill/getAllApplication/getClusterNode/Metrcis
Client->>MRAppMaster: MRClientProtocol:getJob/TaskReport/kill/job/task/taskAttempt
AM 编写, 分成 AM-RM 和 AM-NM 两部分:
例子:
sequenceDiagram
NodeManager->>ResourceManager: 心跳汇报节点信息
ResourceManager->>NodeManager: 心跳返回需释放的 Container等信息
ResourceManager->>ResourceScheduler: NodeUpdate事件.Sheduler分配资源存在内存中
ApplicationMaster->>ResourceScheduler: 心跳领取新分配的 Container
ApplicationMaster->>NodeManager: 分配Container到内部task并启动
# 简化后的 kubernetes 调度逻辑, 为一个优先的 pod 选择最优的 node
# 目前 pod 的排序逻辑只能一个,而 node 的排序则比较丰富
while pod = podQueue.Pop(): # podQueue 的排序是一个扩展点
for node in sort(filter(nodeList): # filter 有多个扩展点;sort 基于 score 也有多个扩展点
bind(pod, node)
# 简化的 yarn ResourceScheduler 调度逻辑, 为一个优先(比如资源由多到少)的 node 选择优先的 queue -> app -> container
# node 的排序逻辑单一,而 queue/app/container 的排序较为细致
for node in sorted(nodeList):
while canStillAssin:
# 这里其实 queue 有层级关系,会从 root 到 leaf 执行, 一直到 App
app = sortedQueues.pop()
# 即 app 里面的 选择一个 container, 会考虑优先级,本地性 等
app.assignContainer(node)
FSAppAttempt.assignContainer(node)
首先对比 NM 的协议 和 CRI
# ContainerManagementProtocol 只看 Stable的
- startContainers
- stopContainers
- getContainerStatuses
# 下面都是 unstable 的
- updateContainer
- signalToContainer
- localize
- reInitializeContainer
- restartContainer
- rollbackLastReInitialization
- commitLastReInitialization
- getLocalizationStatuses
# CRI
- Run/Stop/Remove/ListPodSandbox
- PodSandboxStatus
- Create/Start/Stop/Remove/ListContainer
- ContainerStatus
- UpdateContainerResources
- ExecSync/Exec/Attach/PortForward
原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。
如有侵权,请联系 cloudcommunity@tencent.com 删除。
原创声明:本文系作者授权腾讯云开发者社区发表,未经许可,不得转载。
如有侵权,请联系 cloudcommunity@tencent.com 删除。