YARN底層基礎庫

• Protocol Buffers : Protocol Buffers是Google開源的序列化庫,具有平臺無關,高性能,相容好等優點.Yarn將ProtocolBuffers用到RPC通信中,預設情況下,Yarn RPC中所有參數採用Protocol Buffers進行序列化/反序列化
• Apache Avro : Avro是Hadoop生態系統中的RPC框架,具有平臺無關,支持動態模式等優點,Avro的最初設計動機是解決Yarn RPC相容性和擴展性差等問題
• RPC庫 : Yarn採用MR1中的RPC庫,但其中採用的預設序列化方法被替換成了Protocol Buffers
• 服務庫和事件庫 : Yarn將所有的對象服務化,以便統一管理(創建,銷毀等),而服務之間則採用事件機制進行通信
• 狀態機庫 : 狀態機是一種表示有限個狀態以及在這些狀態之間的轉移和動作等行為的數學模型

• 平臺無關,語言無關
• 高性能,解析速度是XML的20 ~ 100倍
• 體積小,文件大小僅是XML的1/10 ~ 1/3
• 使用簡單
• 相容性好

• 豐富的數據結構類型
• 快速可壓縮的二進位數據形式
• 存儲持久數據的文件容器
• 提供遠程過程調用RPC
• 簡單的動態語言結合功能

• 支持動態模式。Avro不需要生成代碼，有利於搭建通用的數據處理系統，避免了代碼入侵
• 數據無需加標簽
• 無需手工分配的域標識

• 將每個被服務化的對象分為4個狀態 : NOTINITED(被創建)，INITED(已初始化)，STARTED(已啟動)，STOPPED(已停止)
• 任何服務狀態變化都可以觸發另外一些動作
• 可通過組合的方式對任意服務進行組合，以便進行統一管理

 

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;
    }
}
// Task事件類型定義
public enum TaskEventType {
    T_KILL,
    T_SCHEDULE
}

public class JobEvent extends AbstractEvent<JobEventType> {
    private String jobID;
    public JobEvent (String jobID, JobEventType type) {
        super(type);
        this.jobID = jobID;
    }
}
//Job事件類型定義
public enum JobEventType {
    JOB_KILL,
    JOB_INIT,
    JOB_START
}

@SuppressWarnings("unchecked")
public class SimpleMRAppMaster extends CompositeService {
    private Dispatcher dispatcher; //中央非同步調度器
    private String jobID;
    private int taskNumber; //該作業包含的任務數目
    private String[] taskIDs; //該作業內部包含的所有任務
 
    public SimpleMRAppMaster (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] = new String (jobID + "_task_" + i);
        }
    }
 
    public void serviceInit (final Configuration conf) throws Exception {
        dispatcher = new AsyncDispatcher(); //定義一個中央非同步調度器
        //分別註冊Job和Task事件調度器
        dispatcher.register(JobEventType.class, new JobEventDispatcher());
        dispatcher.register(TaskEventType.class, new TaskEventDispatcher());
        addService((Service)dispatcher);
        super.serviceInit(conf);
    }
 
    public Dispatcher getDispatcher() {
        return dispatcher;
    }
 
    private class JobEventDispatcher implements EventHandler<JobEvent> {
        @Override
        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> {
        @Override
        public void handler (TaskEvent event) {
            if (event.getType() == TaskEventType.T_KILL) {
                System.out.println("Receive T_KILL event of task" + event.getTaskID());
            } else if (event.getType() == TaskEventType.T_SCHEDULE) {
                System.out.println("Receive T_SCHEDULE of task" + event.getTaskID()); 
            }
        }
    }
}

@SuppressWarnings("unchecked")
public class SimpleMRAppMasterTest {
    public static void main (String[] args) throws Exception {
        String jobID = "job_20131215_12";
        SimpleMRAppMaster appMaster = new SimpleMRAppMaster("Simple MRAppMaster", jobID, 5);
        YarnConfiguration conf = new YarnConfiguration(new Configuration());
        appMaster.serviceInit(conf);
        appMaster.serviceStart();
        appMaster.getDispatcher().getEventHandler().handle(new JobEvent(jobID, JobEventType.JOB_KILL));
        appMaster.getDispatcher().getEventHandler().handle(new JobEvent(jobID, JobEventType.JOB_INIT));
    }
}