【深入淺出 Yarn 架構與實現】2-3 Yarn 基礎庫
2022-11-10 21:01:10
一個龐大的分散式系統，各個元件間是如何協調工作的？元件是如何解耦的？執行緒執行如何更高效，減少阻塞帶來的低效問題？本節將對 Yarn 的服務庫和事件庫進行介紹，看看 Yarn 是如何解決這些問題的。
一、服務庫
一）簡介
對於生命週期較長的物件，Yarn 採用基於服務的模型對其進行管理，有以下幾個特點：
- 基於狀態管理：分為 4 個狀態：
NOTINITED(被建立)、INITED(已初始化)、STARTED(已啟動)、STOPPED(已停止)。 - 服務狀態的變化會觸發其他的操作。
- 可通過組合的方式對服務進行組合。
二）原始碼簡析
原始碼地址在 hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/service 的 Service 介面中。
其中定義了服務的四個狀態，以及需要實現的狀態轉換、獲取資訊、註冊等方法。
public interface Service extends Closeable {
public enum STATE {
NOTINITED(0, "NOTINITED"),
INITED(1, "INITED"),
STARTED(2, "STARTED"),
STOPPED(3, "STOPPED");
}
void init(Configuration config);
void start();
void stop();
void close() throws IOException;
void registerServiceListener(ServiceStateChangeListener listener);
// ......
抽象類 AbstractService 實現了 Service 介面，提供了基礎的 Service 實現，非組合服務直接繼承這個抽象類再開發即可。
public abstract class AbstractService implements Service {
// 以 start 實現為例，執行後會觸發其他的操作
public void start() {
if (isInState(STATE.STARTED)) {
return;
}
//enter the started state
synchronized (stateChangeLock) {
if (stateModel.enterState(STATE.STARTED) != STATE.STARTED) {
try {
startTime = System.currentTimeMillis();
serviceStart();
if (isInState(STATE.STARTED)) {
//if the service started (and isn't now in a later state), notify
if (LOG.isDebugEnabled()) {
LOG.debug("Service " + getName() + " is started");
}
notifyListeners();
}
} catch (Exception e) {
noteFailure(e);
ServiceOperations.stopQuietly(LOG, this);
throw ServiceStateException.convert(e);
}
}
}
}
// ......
對於組合類的服務如 ResourceManager、NodeManager 等，需要繼承 CompositeService。其中會有對組合服務的邏輯處理。
public List<Service> getServices() {
synchronized (serviceList) {
return new ArrayList<Service>(serviceList);
}
}
protected void addService(Service service) {
if (LOG.isDebugEnabled()) {
LOG.debug("Adding service " + service.getName());
}
synchronized (serviceList) {
serviceList.add(service);
}
}
二、事件庫
傳統函數式呼叫的問題：
整個執行過程是序列、同步進行的。呼叫另一個函數的時候，需要等待函數執行完畢，才會繼續往下走。示意圖如下：
為了解決函數式呼叫的問題，可使用「事件驅動」的程式設計模型。
- 所有物件都被抽象成事件處理器
- 事件處理器之間通過事件相關聯
- 每種事件處理器處理一種事件
- 根據需要會觸發另一種事件
- 每類事件的處理可分割為多個步驟，用有限狀態機表示
- 重要的是有一個「中央非同步排程器（AsyncDispatcher）」，負責對待處理事件的收取和分發
示意圖如下：
通過以上的方式，可以使程式有低耦合高內聚的特點，各個模組僅需完成各自的功能，同時提高了執行效率，把拆分的操作通過事件的方式傳送出去即可。
三、服務庫和事件庫使用案例
本節將實現一個簡化版的 MapReduce ApplicationMaster，幫助瞭解 service 和 event 的使用方法。
與 MR 類似，一個 job 將被分為多個 task 執行。因此涉及 job 和 task 兩種物件的事件。並有一個 AsyncDispatcher 處理排程。
案例已上傳至 github，有幫助可以點個 ⭐️
https://github.com/Simon-Ace/hadoop-yarn-study-demo/tree/master/service-event-demo
一）事件部分
參考 hadoop 原始碼中 Task 和 Job Event 的實現，進行一些簡化。
1、task
public enum TaskEventType {
//Producer:Client, Job
T_KILL,
//Producer:Job
T_SCHEDULE
}
public class TaskEvent extends AbstractEvent<TaskEventType> {
private String taskID;
public TaskEvent(String taskID, TaskEventType type) {
super(type);
this.taskID = taskID;
}
public String getTaskID() {
return taskID;
}
}
2、job
public enum JobEventType {
//Producer:Client
JOB_KILL,
//Producer:MRAppMaster
JOB_INIT
}
public class JobEvent extends AbstractEvent<JobEventType> {
private String jobID;
public JobEvent(String jobID, JobEventType type) {
super(type);
this.jobID = jobID;
}
public String getJobId() {
return jobID;
}
}
二）事件排程器
- 定義和註冊 EventDispatcher
- service 初始化和啟動方法
import com.shuofxz.event.JobEvent;
import com.shuofxz.event.JobEventType;
import com.shuofxz.event.TaskEvent;
import com.shuofxz.event.TaskEventType;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.service.CompositeService;
import org.apache.hadoop.service.Service;
import org.apache.hadoop.yarn.event.AsyncDispatcher;
import org.apache.hadoop.yarn.event.Dispatcher;
import org.apache.hadoop.yarn.event.EventHandler;
@SuppressWarnings("unchecked")
public class MyMRAppMaster extends CompositeService {
private Dispatcher dispatcher; // AsyncDispatcher
private String jobID;
private int taskNumber; // 一個 job 包含的 task 數
private String[] taskIDs;
public MyMRAppMaster(String name, String jobID, int taskNumber) {
super(name);
this.jobID = jobID;
this.taskNumber = taskNumber;
taskIDs = new String[taskNumber];
for (int i = 0; i < taskNumber; i++) {
taskIDs[i] = this.jobID + "_task_" + i;
}
}
public void serviceInit(Configuration conf) throws Exception {
dispatcher = new AsyncDispatcher();
dispatcher.register(JobEventType.class, new JobEventDispatcher()); // register a job
dispatcher.register(TaskEventType.class, new TaskEventDispatcher()); // register a task
addService((Service) dispatcher);
super.serviceInit(conf);
}
public void serviceStart() throws Exception {
super.serviceStart();
}
public Dispatcher getDispatcher() {
return dispatcher;
}
private class JobEventDispatcher implements EventHandler<JobEvent> {
public void handle(JobEvent event) {
if (event.getType() == JobEventType.JOB_KILL) {
System.out.println("Receive JOB_KILL event, killing all the tasks");
for (int i = 0; i < taskNumber; i++) {
dispatcher.getEventHandler().handle(new TaskEvent(taskIDs[i], TaskEventType.T_KILL));
}
} else if (event.getType() == JobEventType.JOB_INIT) {
System.out.println("Receive JOB_INIT event, scheduling tasks");
for (int i = 0; i < taskNumber; i++) {
dispatcher.getEventHandler().handle(new TaskEvent(taskIDs[i], TaskEventType.T_SCHEDULE));
}
}
}
}
private class TaskEventDispatcher implements EventHandler<TaskEvent> {
public void handle(TaskEvent event) {
if (event.getType() == TaskEventType.T_KILL) {
System.out.println("Receive T_KILL event of task id " + event.getTaskID());
} else if (event.getType() == TaskEventType.T_SCHEDULE) {
System.out.println("Receive T_SCHEDULE event of task id " + event.getTaskID());
}
}
}
}
三）測試程式
- 生成一個新的 job
- 觸發事件
JOB_KILL和JOB_INIT
public class MyMRAppMasterTest {
public static void main(String[] args) {
String jobID = "job_20221011_99";
MyMRAppMaster appMaster = new MyMRAppMaster("My MRAppMaster Test", jobID, 10);
YarnConfiguration conf = new YarnConfiguration(new Configuration());
try {
appMaster.serviceInit(conf);
appMaster.serviceStart();
} catch (Exception e) {
e.printStackTrace();
}
appMaster.getDispatcher().getEventHandler().handle(new JobEvent(jobID, JobEventType.JOB_KILL));
appMaster.getDispatcher().getEventHandler().handle(new JobEvent(jobID, JobEventType.JOB_INIT));
}
}
輸出結果：
Receive JOB_KILL event, killing all the tasks
Receive JOB_INIT event, scheduling tasks
Receive T_KILL event of task id job_20150723_11_task_0
Receive T_KILL event of task id job_20150723_11_task_1
Receive T_KILL event of task id job_20150723_11_task_2
Receive T_KILL event of task id job_20150723_11_task_3
Receive T_KILL event of task id job_20150723_11_task_4
Receive T_KILL event of task id job_20150723_11_task_5
Receive T_KILL event of task id job_20150723_11_task_6
Receive T_KILL event of task id job_20150723_11_task_7
Receive T_KILL event of task id job_20150723_11_task_8
Receive T_KILL event of task id job_20150723_11_task_9
Receive T_SCHEDULE event of task id job_20150723_11_task_0
Receive T_SCHEDULE event of task id job_20150723_11_task_1
Receive T_SCHEDULE event of task id job_20150723_11_task_2
Receive T_SCHEDULE event of task id job_20150723_11_task_3
Receive T_SCHEDULE event of task id job_20150723_11_task_4
Receive T_SCHEDULE event of task id job_20150723_11_task_5
Receive T_SCHEDULE event of task id job_20150723_11_task_6
Receive T_SCHEDULE event of task id job_20150723_11_task_7
Receive T_SCHEDULE event of task id job_20150723_11_task_8
Receive T_SCHEDULE event of task id job_20150723_11_task_9
四、總結
本節介紹了 Yarn 的服務和事件庫。
服務庫規範了生命週期較長的服務型物件，定義了服務的四種狀態、啟停註冊等要實現的方法，給出了單一型別和組合型別服務的基本實現。
事件庫的使用，解決了原始函數型呼叫的高耦合、阻塞低效等問題。可將一個大任務拆分成多個小任務，小任務變成不同的事件來觸發處理。每一個事件處理器處理一種事件，並有一箇中央非同步排程器管理事件的收集和分發。
最後用一個簡化的 MR ApplicationMaster 將事件庫和服務庫進行結合，更深體會如何在專案中將其結合使用。
學習過程中，寫一個 demo 能更好的幫助你理解知識。
參考文章：
《Hadoop 技術內幕 - 深入解析 Yarn 結構設計與實現原理》3.4 節