spark之master与worker通信模型讲解

通信模型架构图

master 端代码
import akka.actor.{Actor, ActorSystem, Props}
import com.typesafe.config.ConfigFactory
// 需要导入这2个包 封装一些属性。
class MasterActor extends Actor {

  //在开始之前调用一次
  override def preStart(): Unit = {

  }

  //用于接收消息
  override def receive: Receive = {
    case "started" => {
      println("Master has been started!")
      //进入这个分支,说明这个Master线程已经启动完成
    }
    case "connecting" => {
      println("Master has been get connect from Worker!")
      println("a Worker Node has been register!")
      //返回消息给Worker
      sender() ! "connected"
      Thread.sleep(1000)
    }

    case "stoped" => {

    }
  }

}

object Demo01MasterActor {

  def main(args: Array[String]) {

    //设置MasterIP和端口
    val masterHost = "localhost"
    val masterPort = "1234"

    //端口和IP封装到akka架构,获取一个属性配置文件
    val conStr =
      s"""
         |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
         |akka.remote.netty.tcp.hostname = "$masterHost"
         |akka.remote.netty.tcp.port = "$masterPort"
  """.stripMargin

    val config = ConfigFactory.parseString(conStr)
    val masterActorSystem = ActorSystem("MasterActorSystem", config)
    val masterActor = masterActorSystem.actorOf(Props[MasterActor], "MasterActor")
    masterActor ! "started"
    masterActorSystem.awaitTermination();

  }

}

worker端代码
import akka.actor.{Actor, ActorSelection, ActorSystem, Props}
import com.typesafe.config.ConfigFactory

class WorkerActor extends Actor {
  var masterURL: ActorSelection = null

  //启动Actor之前执行,做初始化工作
  override def preStart(): Unit = {
    //配置访问Master的URL
    //MasterIP:localhost
    //MasterPort:8888(根据Master配置)
    //Master的 ActorSystem对象:MasterActorSystem、MasterActor
    masterURL = context.actorSelection("akka.tcp://[email protected]:8888/user/MasterActor")
  }

  override def receive: Receive = {
    case "started" => {
      println("Worker has been started!")
      //进入这个分支,说明这个Worker线程已经启动完成
      //可以去向Master注册

      //请求和Master建立连接
      masterURL ! "connecting"
    }
    case "connected" => {
      println("Worker 收到来自Master确认信息!")
    }
    case "stoped" => {

    }
  }

}

object Demo01WorkerActor {

  def main(args: Array[String]) {
    //初始化MastereIP和端口、WorkerIP和端口

    //    val masterHost = args(0)
    //    val masterPort = args(1)
    //    val workerHost = args(2)
    //    val workePort = args(3)

    val masterHost = "localhost"
    val masterPort = "8888"

    val workerHost = "localhost"
    val workePort = "8889"

    //端口和IP封装到akka架构,获取一个属性配置文件
    val conStr =
      s"""
         |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
         |akka.remote.netty.tcp.hostname = "$workerHost"
         |akka.remote.netty.tcp.port = "$workePort"
  """.stripMargin

    val config = ConfigFactory.parseString(conStr)
    val workerActorSystem = ActorSystem("WorkerActorSystem", config)
    val workerActor = workerActorSystem.actorOf(Props[WorkerActor], "WorkerActor")
    workerActor ! "started"
    workerActorSystem.awaitTermination();

  }

}
时间: 2024-10-08 10:04:26

spark之master与worker通信模型讲解的相关文章

Spark的Master和Worker集群启动的源码分析

基于spark1.3.1的源码进行分析 spark master启动源码分析 1.在start-master.sh调用master的main方法,main方法调用 def main(argStrings: Array[String]) { SignalLogger.register(log) val conf = new SparkConf val args = new MasterArguments(argStrings, conf) val (actorSystem, _, _, _) =

Spark技术内幕:Client,Master和Worker 通信源码解析

Spark的Cluster Manager可以有几种部署模式: Standlone Mesos YARN EC2 Local 在向集群提交计算任务后,系统的运算模型就是Driver Program定义的SparkContext向APP Master提交,有APP Master进行计算资源的调度并最终完成计算.具体阐述可以阅读<Spark:大数据的电花火石!>. 那么Standalone模式下,Client,Master和Worker是如何进行通信,注册并开启服务的呢? 1. node之间的IP

Spark技术内幕:Client,Master和Worker 通信源代码解析

Spark的Cluster Manager能够有几种部署模式: Standlone Mesos YARN EC2 Local 在向集群提交计算任务后,系统的运算模型就是Driver Program定义的SparkContext向APP Master提交,有APP Master进行计算资源的调度并终于完毕计算.具体阐述能够阅读<Spark:大数据的电花火石!>. 那么Standalone模式下,Client.Master和Worker是怎样进行通信,注冊并开启服务的呢? 1. node之间的RP

Spark分析之Master、Worker以及Application三者之间如何建立连接

Master.preStart(){ webUi.bind() context.system.scheduler.schedule(0 millis, WORKER_TIMEOUT millis, self, CheckForWorkerTimeOut) //定时任务检测是否有DEAD WORKER需要移除 case CheckForWorkerTimeOut => { timeOutDeadWorkers() } /** Check for, and remove, any timed-out

spark 源码理解2 进一步窥探Master、Worker通信机制

上一篇文章 spark 源码理解1 从spark启动脚本开始 是分析执行start_all.sh时,集群中启动了哪些进程,下面我们再深入一点看看这些进程都是做什么用的,它们之间又是如何通信的? 一.Master进程的启动 Master进程,它主要负责对Worker.Driver.App等资源的管理并与它们进行通信,这篇文章中我打算着重讲一下它与Worker的通信,其它的部分放在以后的章节再加以描述. spark-daemon.sh start org.apache.spark.deploy.ma

SPARK的MAster资源调度原理(源码)分析

SPARK的MAster资源分配算法(SPARK1.3) master资调度通过源码中的 org.apache.spark.deploy.master包下的schedule()方法实现 步骤如下: 首先判断master是否是alive状态,如果不是alive则返回,也就是只有活动的master才会进行资源调度,standby master是不会进行资源调度的 把之前注册的worker中的alive状态的worker传入 Random.shuffer方法,该方法主要是把worker顺序打乱,返回一

Spark技术内幕:Worker源码与架构解析

首先通过一张Spark的架构图来了解Worker在Spark中的作用和地位: Worker所起的作用有以下几个: 1. 接受Master的指令,启动或者杀掉Executor 2. 接受Master的指令,启动或者杀掉Driver 3. 报告Executor/Driver的状态到Master 4. 心跳到Master,心跳超时则Master认为Worker已经挂了不能工作了 5. 向GUI报告Worker的状态 说白了,Worker就是整个集群真正干活的.首先看一下Worker重要的数据结构: v

nginx源码分析--master和worker进程模型

一.Nginx整体架构 正常执行中的nginx会有多个进程,最基本的有master process(监控进程,也叫做主进程)和woker process(工作进程),还可能有cache相关进程. 一个较为完整的整体框架结构如图所示: 二.核心进程模型 启动nginx的主进程将充当监控进程,而由主进程fork()出来的子进程则充当工作进程. nginx也可以单进程模型执行,在这种进程模型下,主进程就是工作进程,没有监控进程. Nginx的核心进程模型框图如下: master进程 监控进程充当整个进

Master和worker模式

让和hadoop的设计思想是一样的,Master负责分配任务和获取任务的结果,worker是真正处理业务逻辑的. 使用ConcurrentLikedQueue去承载所有的任务,因为会有多个worker会并发修改这个队列. public class Task { private int id; private int price ; public int getId() { return id; } public void setId(int id) { this.id = id; } publi