在上一篇文章中 ，我們了解了mesos是什么，它如何有用，并開始使用它。 在本文中，我們將看到如何在mesos上編寫自己的框架。 （在mesos中，框架是在其上運(yùn)行的任何應(yīng)用程序。）本文介紹了一個(gè)名為“ mesos-pinspider”的框架，該框架獲取用戶的pinterest頁面的用戶配置文件信息和用戶面板信息。

Mesos框架

通常，Mesos框架具有三個(gè)基本組件。

• 將任務(wù)提交給框架的驅(qū)動(dòng)程序
• 向主服務(wù)器注冊要提供資源的調(diào)度程序 ，執(zhí)行任務(wù)并在執(zhí)行程序上運(yùn)行它們
• 在從屬節(jié)點(diǎn)上啟動(dòng)以執(zhí)行框架任務(wù)的執(zhí)行程序進(jìn)程

Pinspider框架示例

您可以在github上檢查代碼。 讓我們將其細(xì)分為PinDriver，PinScheduler和Pin UserProfileExecutor。

司機(jī)

該框架的驅(qū)動(dòng)程序組件是PinDriver。

• 創(chuàng)建執(zhí)行人信息

  使用Builder模式描述有關(guān)執(zhí)行程序的信息，而mesos使用Google協(xié)議緩沖區(qū)進(jìn)行數(shù)據(jù)交換。 在這里，我們需要設(shè)置executorID，該命令基本上是一個(gè)shell命令，通過以下命令執(zhí)行：'/ bin / sh -c value'。 在執(zhí)行命令之前，將獲取指定的所有URI。 名稱由setName（）設(shè)置。 來源由
  setSource（），框架用來跟蹤執(zhí)行程序的源的標(biāo)識(shí)符樣式字符串。 當(dāng)不同的執(zhí)行者ID可能在語義上相關(guān)時(shí)，這很有用。

  Protos.ExecutorInfo userProfileExecutorInfo = Protos.ExecutorInfo.newBuilder().setExecutorId(Protos.ExecutorID.newBuilder().setValue("PinUserProfileExecutor")).setCommand(commandInfoUserProfile).setName("PinUserProfileExecutor Java").setSource("java").build();
• 創(chuàng)建框架信息

  描述框架信息。 用戶字段用于確定執(zhí)行程序/任務(wù)應(yīng)以其啟動(dòng)的Unix用戶。 如果用戶字段設(shè)置為空字符串，Mesos將自動(dòng)將其設(shè)置為當(dāng)前用戶。 主機(jī)在刪除框架之前等待調(diào)度程序進(jìn)行故障轉(zhuǎn)移的時(shí)間由以下方式指定：
  setFailoverTimeout（）。 框架的名稱由setName（）設(shè)置

  Protos.FrameworkInfo.Builder frameworkBuilder = Protos.FrameworkInfo.newBuilder().setFailoverTimeout(120000).setUser("").setName("Pinspider Framework");
• 實(shí)例化調(diào)度程序

  您需要使用要提交的執(zhí)行程序?qū)嵗蝿?wù)數(shù)量來實(shí)例化調(diào)度程序。

  Scheduler scheduler = args.length == 1 ?new PinScheduler(userProfileExecutorInfo,userBoardExecutorInfo) :new PinScheduler(userProfileExecutorInfo, userBoardExecutorInfo, Integer.parseInt(args[1]), args[2]);

  注意：請注意，使用了兩個(gè)ExecutorInfo。 一個(gè)用于獲取用戶配置文件信息，另一個(gè)用于獲取用戶板信息以進(jìn)行演示。 此說明僅涉及一個(gè)executorinfo – userProfileExecutorInfo

• 啟動(dòng)mesos調(diào)度程序驅(qū)動(dòng)程序。

  MesosSchedulerDriver是SchedulerDriver的實(shí)現(xiàn)，SchedulerDriver是將調(diào)度程序連接到mesos的抽象接口。 這是通過管理調(diào)度程序的生命周期（啟動(dòng)，停止和等待任務(wù)完成）以及與Mesos交互（啟動(dòng)任務(wù)，終止任務(wù)等）來完成的。

  MesosSchedulerDriver schedulerDriver =new MesosSchedulerDriver(scheduler,frameworkBuilder.build(), args[0]);int status = schedulerDriver.run() == Protos.Status.DRIVER_STOPPED ? 0 : 1;schedulerDriver.stop();System.exit(status);

執(zhí)行器執(zhí)行

框架的執(zhí)行器組件是PinUserProfileExecutor。

執(zhí)行程序是由框架的執(zhí)行程序?qū)崿F(xiàn)的回調(diào)接口。 在我們的實(shí)現(xiàn)中，讓我們專注于launchTask（）

@Override public void launchTask(final ExecutorDriver executorDriver final Protos.TaskInfo taskInfo) { }

• 通過使用構(gòu)建器模式設(shè)置ID和狀態(tài)來設(shè)置任務(wù)狀態(tài)。 Protos.TaskStatus taskStatus =Protos.TaskStatus.newBuilder().setTaskId(taskInfo.getTaskId()) .setState(Protos.TaskState.TASK_RUNNING).build();
• 將狀態(tài)更新發(fā)送到框架調(diào)度程序，根據(jù)需要重試，直到收到確認(rèn)或執(zhí)行程序終止為止，在這種情況下，將發(fā)送TASK_LOST狀態(tài)更新。 executorDriver.sendStatusUpdate(taskStatus);
• 從任務(wù)中獲取數(shù)據(jù)并運(yùn)行邏輯。 try {message = ("userprofile :" + getUserProfileInfo(url)).getBytes(); } catch (IOException e) {LOGGER.error("Error parsing the Pinterest URL :" + e.getMessage()); }
• 向框架發(fā)送消息。 executorDriver.sendFrameworkMessage(message);
• 將任務(wù)的狀態(tài)標(biāo)記為已完成，然后將狀態(tài)更新發(fā)送到框架調(diào)度程序。 taskStatus = Protos.TaskStatus.newBuilder().setTaskId(taskInfo.getTaskId()).setState(Protos.TaskState.TASK_FINISHED).build(); executorDriver.sendStatusUpdate(taskStatus);
• main（）方法創(chuàng)建MesosExecutorDriver實(shí)例并運(yùn)行 mesosExecutorDriver.run() == Protos.Status.DRIVER_STOPPED ? 0 : 1

調(diào)度程序?qū)嵤?/h2>

該框架的Scheduler組件是Pin Scheduler。

調(diào)度程序是由框架的調(diào)度程序?qū)崿F(xiàn)的回調(diào)接口。 在我們的實(shí)現(xiàn)中，讓我們專注于resourceOffers（），statusUpdate（）和frameworkMessage（）

• 構(gòu)造函數(shù)：使用執(zhí)行程序信息和啟動(dòng)任務(wù)數(shù)進(jìn)行構(gòu)造。 public PinScheduler(Protos.ExecutorInfo pinUserProfileExecutor , Protos.ExecutorInfo pinUserBoardExecutor ) {this(pinUserProfileExecutor,pinUserBoardExecutor, 5, "http://www.pinterest.com/techcrunch"); } public PinScheduler(Protos.ExecutorInfo pinUserProfileExecutor,Protos.ExecutorInfo pinUserBoardExecutor, int totalTasks, String url) { this.pinUserProfileExecutor = pinUserProfileExecutor;this.pinUserBoardExecutor = pinUserBoardExecutor;this.totalTasks = totalTasks; this.crawlQueue =Collections.synchronizedList(new ArrayList<String>());this.crawlQueue.add(url); }
• 資源報(bào)價(jià)
  • 資源商品可以是CPU，內(nèi)存等資源。從商品列表中，獲取資源的標(biāo)量值。 在設(shè)置任務(wù)信息時(shí)，我們需要提供任務(wù)資源的需求。 for (Protos.Offer offer : list) {List<Protos.TaskInfo> taskInfoList = new ArrayList<Protos.TaskInfo>();double offerCpus = 0; double offerMem = 0;for (Protos.Resource resource : offer.getResourcesList()) {if (resource.getName().equals("cpus")) {offerCpus += resource.getScalar().getValue();}else if (resource.getName().equals("mem")) {offerMem += resource.getScalar().getValue();}}LOGGER.info("Received Offer : " + offer.getId().getValue() +" with cpus = " + offerCpus + " and mem =" + offerMem);
  • 創(chuàng)建任務(wù)ID。 Protos.TaskID taskID = Protos.TaskID.newBuilder().setValue(Integer.toString(launchedTasks++)).build();
  • 通過設(shè)置任務(wù)ID，添加資源，設(shè)置數(shù)據(jù)和設(shè)置執(zhí)行程序來創(chuàng)建任務(wù)信息。 Protos.TaskInfo pinUserProfileTaskInfo = Protos.TaskInfo.newBuilder().setName("task " + taskID.getValue()).setTaskId(taskID).setSlaveId(offer.getSlaveId()).addResources(Protos.Resource.newBuilder().setName("cpus").setType(Protos.Value.Type.SCALAR).setScalar(Protos.Value.Scalar.newBuilder().setValue(CPUS_PER_TASK))).addResources(Protos.Resource.newBuilder().setName("mem").setType(Protos.Value.Type.SCALAR).setScalar(Protos.Value.Scalar.newBuilder().setValue(MEM_PER_TASK))).setData(ByteString.copyFromUtf8(crawlQueue.get(0))).setExecutor(Protos.ExecutorInfo.newBuilder(pinUserProfileExecutor)).build();
  • 通過SchedulerDriver啟動(dòng)任務(wù)。 ... taskInfoList.add(pinUserProfileTaskInfo); taskInfoList.add(pinUserBoardTaskInfo); } schedulerDriver.launchTasks(offer.getId(), taskInfoList);
• 狀態(tài)更新

  當(dāng)任務(wù)的狀態(tài)已更改（即，從屬丟失且任務(wù)丟失），任務(wù)完成且執(zhí)行者發(fā)送狀態(tài)更新時(shí)，調(diào)用此方法。

  @Override public void statusUpdate(SchedulerDriver schedulerDriver, Protos.TaskStatus taskStatus) { ... }
• • 如果任務(wù)完成，請停止SchedulerDriver if (taskStatus.getState() == Protos.TaskState.TASK_FINISHED) {finishedTasks++;LOGGER.info("Finished tasks : " + finishedTasks);if (finishedTasks == totalTasks) {schedulerDriver.stop();}}
  • 如果任務(wù)被殺死，丟失或失敗，則中止SchedulerDriver if (taskStatus.getState() == Protos.TaskState.TASK_FAILED || taskStatus.getState() == Protos.TaskState.TASK_KILLED || taskStatus.getState() == Protos.TaskState.TASK_LOST) {LOGGER.error("Aborting because the task " + taskStatus.getTaskId().getValue() +" is in unexpected state : " + taskStatus.getState().getValueDescriptor().getName() +"with reason : " + taskStatus.getReason().getValueDescriptor().getName()+ " from source : " + taskStatus.getSource().getValueDescriptor().getName() + " with message : " + taskStatus.getMessage());schedulerDriver.abort(); }
• 框架訊息

  當(dāng)執(zhí)行程序發(fā)送消息時(shí)，將調(diào)用此函數(shù)。

• • 處理您的訊息 @Override public void frameworkMessage(SchedulerDriver schedulerDriver, Protos.ExecutorID executorID, Protos.SlaveID slaveID, byte[] bytes) {String data = new String(bytes);System.out.println(data);LOGGER.info("User Profile Information : " + data); }

此處提供完整的代碼，以及運(yùn)行和輸出示例的說明。

翻譯自: https://www.javacodegeeks.com/2015/01/apache-mesos-writing-your-own-distributed-frameworks.html

總結(jié)

以上是生活随笔為你收集整理的Apache Mesos：编写您自己的分布式框架的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。