spark启动原理总结

 1  override def onStart(): Unit = {
 2     logInfo("Starting Spark master at " + masterUrl)
 3     logInfo(s"Running Spark version ${org.apache.spark.SPARK_VERSION}")
 4     webUi = new MasterWebUI(this, webUiPort)
 5     webUi.bind()
 6     masterWebUiUrl = "http://" + masterPublicAddress + ":" + webUi.boundPort
 7     if (reverseProxy) {
 8       masterWebUiUrl = conf.get("spark.ui.reverseProxyUrl", masterWebUiUrl)
 9       webUi.addProxy()
10       logInfo(s"Spark Master is acting as a reverse proxy. Master, Workers and " +
11        s"Applications UIs are available at $masterWebUiUrl")
12     }
13     checkForWorkerTimeOutTask = forwardMessageThread.scheduleAtFixedRate(new Runnable {
14       override def run(): Unit = Utils.tryLogNonFatalError {
15         self.send(CheckForWorkerTimeOut)
16       }
17     }, 0, WORKER_TIMEOUT_MS, TimeUnit.MILLISECONDS)
18 
19     if (restServerEnabled) {
20       val port = conf.getInt("spark.master.rest.port", 6066)
21       restServer = Some(new StandaloneRestServer(address.host, port, conf, self, masterUrl))
22     }
23     restServerBoundPort = restServer.map(_.start())
24 
25     masterMetricsSystem.registerSource(masterSource)
26     masterMetricsSystem.start()
27     applicationMetricsSystem.start()
28     // Attach the master and app metrics servlet handler to the web ui after the metrics systems are
29     // started.
30     masterMetricsSystem.getServletHandlers.foreach(webUi.attachHandler)
31     applicationMetricsSystem.getServletHandlers.foreach(webUi.attachHandler)
32 
33     val serializer = new JavaSerializer(conf)
34     val (persistenceEngine_, leaderElectionAgent_) = RECOVERY_MODE match {
35       case "ZOOKEEPER" =>
36         logInfo("Persisting recovery state to ZooKeeper")
37         val zkFactory =
38           new ZooKeeperRecoveryModeFactory(conf, serializer)
39         (zkFactory.createPersistenceEngine(), zkFactory.createLeaderElectionAgent(this))
40       case "FILESYSTEM" =>
41         val fsFactory =
42           new FileSystemRecoveryModeFactory(conf, serializer)
43         (fsFactory.createPersistenceEngine(), fsFactory.createLeaderElectionAgent(this))
44       case "CUSTOM" =>
45         val clazz = Utils.classForName(conf.get("spark.deploy.recoveryMode.factory"))
46         val factory = clazz.getConstructor(classOf[SparkConf], classOf[Serializer])
47           .newInstance(conf, serializer)
48           .asInstanceOf[StandaloneRecoveryModeFactory]
49         (factory.createPersistenceEngine(), factory.createLeaderElectionAgent(this))
50       case _ =>
51         (new BlackHolePersistenceEngine(), new MonarchyLeaderAgent(this))
52     }
53     persistenceEngine = persistenceEngine_
54     leaderElectionAgent = leaderElectionAgent_
55   }