spark storage之SparkEnv

 1   /**
 2    * Create a SparkEnv for the driver.
 3    */
 4   private[spark] def createDriverEnv(
 5       conf: SparkConf,
 6       isLocal: Boolean,
 7       listenerBus: LiveListenerBus,
 8       mockOutputCommitCoordinator: Option[OutputCommitCoordinator] = None): SparkEnv = {
 9     assert(conf.contains("spark.driver.host"), "spark.driver.host is not set on the driver!")
10     assert(conf.contains("spark.driver.port"), "spark.driver.port is not set on the driver!")
11     val hostname = conf.get("spark.driver.host")
12     val port = conf.get("spark.driver.port").toInt
13     create(
14       conf,
15       SparkContext.DRIVER_IDENTIFIER,
16       hostname,
17       port,
18       isDriver = true,
19       isLocal = isLocal,
20       listenerBus = listenerBus,
21       mockOutputCommitCoordinator = mockOutputCommitCoordinator
22     )
23   }

  1   private def create(
  2       conf: SparkConf,
  3       executorId: String,
  4       hostname: String,
  5       port: Int,
  6       isDriver: Boolean,
  7       isLocal: Boolean,
  8       listenerBus: LiveListenerBus = null,
  9       numUsableCores: Int = 0,
 10       mockOutputCommitCoordinator: Option[OutputCommitCoordinator] = None): SparkEnv = {
 11 
 12     // Listener bus is only used on the driver
 13     if (isDriver) {
 14       assert(listenerBus != null, "Attempted to create driver SparkEnv with null listener bus!")
 15     }
 16 
 17     val securityManager = new SecurityManager(conf)
 18 
 19     // Create the ActorSystem for Akka and get the port it binds to.
 20     val actorSystemName = if (isDriver) driverActorSystemName else executorActorSystemName
 21     val rpcEnv = RpcEnv.create(actorSystemName, hostname, port, conf, securityManager)
 22     val actorSystem = rpcEnv.asInstanceOf[AkkaRpcEnv].actorSystem
 23 
 24     // Figure out which port Akka actually bound to in case the original port is 0 or occupied.
 25     if (isDriver) {
 26       conf.set("spark.driver.port", rpcEnv.address.port.toString)
 27     } else {
 28       conf.set("spark.executor.port", rpcEnv.address.port.toString)
 29     }
 30 
 31     // Create an instance of the class with the given name, possibly initializing it with our conf
 32     def instantiateClass[T](className: String): T = {
 33       val cls = Utils.classForName(className)
 34       // Look for a constructor taking a SparkConf and a boolean isDriver, then one taking just
 35       // SparkConf, then one taking no arguments
 36       try {
 37         cls.getConstructor(classOf[SparkConf], java.lang.Boolean.TYPE)
 38           .newInstance(conf, new java.lang.Boolean(isDriver))
 39           .asInstanceOf[T]
 40       } catch {
 41         case _: NoSuchMethodException =>
 42           try {
 43             cls.getConstructor(classOf[SparkConf]).newInstance(conf).asInstanceOf[T]
 44           } catch {
 45             case _: NoSuchMethodException =>
 46               cls.getConstructor().newInstance().asInstanceOf[T]
 47           }
 48       }
 49     }
 50 
 51     // Create an instance of the class named by the given SparkConf property, or defaultClassName
 52     // if the property is not set, possibly initializing it with our conf
 53     def instantiateClassFromConf[T](propertyName: String, defaultClassName: String): T = {
 54       instantiateClass[T](conf.get(propertyName, defaultClassName))
 55     }
 56 
 57     val serializer = instantiateClassFromConf[Serializer](
 58       "spark.serializer", "org.apache.spark.serializer.JavaSerializer")
 59     logDebug(s"Using serializer: ${serializer.getClass}")
 60 
 61     val closureSerializer = instantiateClassFromConf[Serializer](
 62       "spark.closure.serializer", "org.apache.spark.serializer.JavaSerializer")
 63 
 64     def registerOrLookupEndpoint(
 65         name: String, endpointCreator: => RpcEndpoint):
 66       RpcEndpointRef = {
 67       if (isDriver) {
 68         logInfo("Registering " + name)
 69         rpcEnv.setupEndpoint(name, endpointCreator)
 70       } else {
 71         RpcUtils.makeDriverRef(name, conf, rpcEnv)
 72       }
 73     }
 74 
 75     val mapOutputTracker = if (isDriver) {
 76       new MapOutputTrackerMaster(conf)
 77     } else {
 78       new MapOutputTrackerWorker(conf)
 79     }
 80 
 81     // Have to assign trackerActor after initialization as MapOutputTrackerActor
 82     // requires the MapOutputTracker itself
 83     mapOutputTracker.trackerEndpoint = registerOrLookupEndpoint(MapOutputTracker.ENDPOINT_NAME,
 84       new MapOutputTrackerMasterEndpoint(
 85         rpcEnv, mapOutputTracker.asInstanceOf[MapOutputTrackerMaster], conf))
 86 
 87     // Let the user specify short names for shuffle managers
 88     val shortShuffleMgrNames = Map(
 89       "hash" -> "org.apache.spark.shuffle.hash.HashShuffleManager",
 90       "sort" -> "org.apache.spark.shuffle.sort.SortShuffleManager",
 91       "tungsten-sort" -> "org.apache.spark.shuffle.unsafe.UnsafeShuffleManager")
 92     val shuffleMgrName = conf.get("spark.shuffle.manager", "sort")
 93     val shuffleMgrClass = shortShuffleMgrNames.getOrElse(shuffleMgrName.toLowerCase, shuffleMgrName)
 94     val shuffleManager = instantiateClass[ShuffleManager](shuffleMgrClass)
 95 
 96     val shuffleMemoryManager = ShuffleMemoryManager.create(conf, numUsableCores)
 97 
 98     val blockTransferService =
 99       conf.get("spark.shuffle.blockTransferService", "netty").toLowerCase match {
100         case "netty" =>
101           new NettyBlockTransferService(conf, securityManager, numUsableCores)
102         case "nio" =>
103           logWarning("NIO-based block transfer service is deprecated, " +
104             "and will be removed in Spark 1.6.0.")
105           new NioBlockTransferService(conf, securityManager)
106       }
107 
108     val blockManagerMaster = new BlockManagerMaster(registerOrLookupEndpoint(
109       BlockManagerMaster.DRIVER_ENDPOINT_NAME,
110       new BlockManagerMasterEndpoint(rpcEnv, isLocal, conf, listenerBus)),
111       conf, isDriver)
112 
113     // NB: blockManager is not valid until initialize() is called later.
114     val blockManager = new BlockManager(executorId, rpcEnv, blockManagerMaster,
115       serializer, conf, mapOutputTracker, shuffleManager, blockTransferService, securityManager,
116       numUsableCores)
117 
118     val broadcastManager = new BroadcastManager(isDriver, conf, securityManager)
119 
120     val cacheManager = new CacheManager(blockManager)
121 
122     val httpFileServer =
123       if (isDriver) {
124         val fileServerPort = conf.getInt("spark.fileserver.port", 0)
125         val server = new HttpFileServer(conf, securityManager, fileServerPort)
126         server.initialize()
127         conf.set("spark.fileserver.uri", server.serverUri)
128         server
129       } else {
130         null
131       }
132 
133     val metricsSystem = if (isDriver) {
134       // Don‘t start metrics system right now for Driver.
135       // We need to wait for the task scheduler to give us an app ID.
136       // Then we can start the metrics system.
137       MetricsSystem.createMetricsSystem("driver", conf, securityManager)
138     } else {
139       // We need to set the executor ID before the MetricsSystem is created because sources and
140       // sinks specified in the metrics configuration file will want to incorporate this executor‘s
141       // ID into the metrics they report.
142       conf.set("spark.executor.id", executorId)
143       val ms = MetricsSystem.createMetricsSystem("executor", conf, securityManager)
144       ms.start()
145       ms
146     }
147 
148     // Set the sparkFiles directory, used when downloading dependencies.  In local mode,
149     // this is a temporary directory; in distributed mode, this is the executor‘s current working
150     // directory.
151     val sparkFilesDir: String = if (isDriver) {
152       Utils.createTempDir(Utils.getLocalDir(conf), "userFiles").getAbsolutePath
153     } else {
154       "."
155     }
156 
157     val outputCommitCoordinator = mockOutputCommitCoordinator.getOrElse {
158       new OutputCommitCoordinator(conf, isDriver)
159     }
160     val outputCommitCoordinatorRef = registerOrLookupEndpoint("OutputCommitCoordinator",
161       new OutputCommitCoordinatorEndpoint(rpcEnv, outputCommitCoordinator))
162     outputCommitCoordinator.coordinatorRef = Some(outputCommitCoordinatorRef)
163 
164     val executorMemoryManager: ExecutorMemoryManager = {
165       val allocator = if (conf.getBoolean("spark.unsafe.offHeap", false)) {
166         MemoryAllocator.UNSAFE
167       } else {
168         MemoryAllocator.HEAP
169       }
170       new ExecutorMemoryManager(allocator)
171     }
172 
173     val envInstance = new SparkEnv(
174       executorId,
175       rpcEnv,
176       serializer,
177       closureSerializer,
178       cacheManager,
179       mapOutputTracker,
180       shuffleManager,
181       broadcastManager,
182       blockTransferService,
183       blockManager,
184       securityManager,
185       httpFileServer,
186       sparkFilesDir,
187       metricsSystem,
188       shuffleMemoryManager,
189       executorMemoryManager,
190       outputCommitCoordinator,
191       conf)
192 
193     // Add a reference to tmp dir created by driver, we will delete this tmp dir when stop() is
194     // called, and we only need to do it for driver. Because driver may run as a service, and if we
195     // don‘t delete this tmp dir when sc is stopped, then will create too many tmp dirs.
196     if (isDriver) {
197       envInstance.driverTmpDirToDelete = Some(sparkFilesDir)
198     }
199 
200     envInstance
201   }

1 val rpcEnv = RpcEnv.create(actorSystemName, hostname, port, conf, securityManager)

 1   private def getRpcEnvFactory(conf: SparkConf): RpcEnvFactory = {
 2     // Add more RpcEnv implementations here
 3     val rpcEnvNames = Map("akka" -> "org.apache.spark.rpc.akka.AkkaRpcEnvFactory")
 4     val rpcEnvName = conf.get("spark.rpc", "akka")
 5     val rpcEnvFactoryClassName = rpcEnvNames.getOrElse(rpcEnvName.toLowerCase, rpcEnvName)
 6     Utils.classForName(rpcEnvFactoryClassName).newInstance().asInstanceOf[RpcEnvFactory]
 7   }
 8 
 9   def create(
10       name: String,
11       host: String,
12       port: Int,
13       conf: SparkConf,
14       securityManager: SecurityManager): RpcEnv = {
15     // Using Reflection to create the RpcEnv to avoid to depend on Akka directly
16     val config = RpcEnvConfig(conf, name, host, port, securityManager)
17     getRpcEnvFactory(conf).create(config)
18   }

1 private[spark] class AkkaRpcEnvFactory extends RpcEnvFactory {
2 
3   def create(config: RpcEnvConfig): RpcEnv = {
4     val (actorSystem, boundPort) = AkkaUtils.createActorSystem(
5       config.name, config.host, config.port, config.conf, config.securityManager)
6     actorSystem.actorOf(Props(classOf[ErrorMonitor]), "ErrorMonitor")
7     new AkkaRpcEnv(actorSystem, config.conf, boundPort)
8   }
9 }

 1     def registerOrLookupEndpoint(
 2         name: String, endpointCreator: => RpcEndpoint):
 3       RpcEndpointRef = {
 4       if (isDriver) {
 5         logInfo("Registering " + name)
 6         rpcEnv.setupEndpoint(name, endpointCreator)
 7       } else {
 8         RpcUtils.makeDriverRef(name, conf, rpcEnv)
 9       }
10     }

1   def makeDriverRef(name: String, conf: SparkConf, rpcEnv: RpcEnv): RpcEndpointRef = {
2     val driverActorSystemName = SparkEnv.driverActorSystemName
3     val driverHost: String = conf.get("spark.driver.host", "localhost")
4     val driverPort: Int = conf.getInt("spark.driver.port", 7077)
5     Utils.checkHost(driverHost, "Expected hostname")
6     rpcEnv.setupEndpointRef(driverActorSystemName, RpcAddress(driverHost, driverPort), name)
7   }

 1     // Create and start the scheduler
 2     val (sched, ts) = SparkContext.createTaskScheduler(this, master)
 3     _schedulerBackend = sched
 4     _taskScheduler = ts
 5     _dagScheduler = new DAGScheduler(this)
 6     _heartbeatReceiver.ask[Boolean](TaskSchedulerIsSet)
 7 
 8     // start TaskScheduler after taskScheduler sets DAGScheduler reference in DAGScheduler‘s
 9     // constructor
10     _taskScheduler.start()