Explanation of master and worker Communication Model of spark

Communication model architecture diagram

The master side code import akka.actor. {Actor, ActorSystem, Props} import com.typesafe.config.ConfigFactory// needs to import these two packages to encapsulate some properties. Class MasterActor extends Actor {/ / calls override def preStart (): Unit = {} / / to receive the message override def receive: Receive = {case "started" = > {println ("Master has been started!") / / before starting, indicating that the Master thread has started and completed} case "connecting" = > {println ("Master has been get connect from Worker!") Println ("a Worker Node has been register!") / / returns a message to Worker sender ()! "connected" Thread.sleep (1000)} case "stoped" = > {} object Demo01MasterActor {def main (args: Array [String]) {/ / set MasterIP and port val masterHost = "localhost" val masterPort = "1234" / / Port and IP are encapsulated into akka architecture Get an attribute profile 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 side code import akka.actor. {Actor, ActorSelection, ActorSystem, Props} import com.typesafe.config.ConfigFactoryclass WorkerActor extends Actor {var masterURL: ActorSelection = null / / execute before starting Actor Do initialization work override def preStart (): Unit = {/ / configure URL / / MasterIP:localhost / / MasterPort:8888 (according to Master configuration) / / ActorSystem object of Master to access Master: MasterActorSystem, MasterActor masterURL = context.actorSelection ("akka.tcp://MasterActorSystem@localhost:8888/user/MasterActor")} override def receive: Receive = {case "started" = > {println ("Worker has been started!") / / enter this branch Indicates that this Worker thread has started and / / can register / / request with Master to establish a connection with Master masterURL! "connecting"} case "connected" = > {println ("Worker received confirmation message from Master!")} case "stoped" = > {} object Demo01WorkerActor {def main (args: Array [String]) {/ / initialize MastereIP and port, WorkerIP and port / / 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" / / Port and IP encapsulated into akka architecture Get an attribute profile 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 ();}}