《深入理解Spark-核心思想与源码分析》读书笔记（1）

    class SparkContext(config: SparkConf) extends Logging with ExecutorAllocationClient

            //Set our own authenticator鉴别器 to properly专有的 negotiate协商
            //userpassword for HTTP connections. This is needed by the HTTP client
            //fetching from the HttpServer. Put here so its only set once.
              if (authOn) {
                Authenticator.setDefault(
                  new Authenticator() {
                override def getPasswordAuthentication(): PasswordAuthentication = {
                  var passAuth: PasswordAuthentication = null
                  val userInfo = getRequestingURL().getUserInfo()
                  if (userInfo != null) {
                    val  parts = userInfo.split(":", 2)
                    passAuth = new PasswordAuthentication(parts(0), parts(1).toCharArray())
                }
                  return passAuth
                }
                  }
                )
              }

def initialize() {
    baseDir = Utils.createTempDir(Utils.getLocalDir(conf), "httpd")
    fileDir = new File(baseDir, "files")
    jarDir = new File(baseDir, "jars")
    fileDir.mkdir()
    jarDir.mkdir()
    logInfo("HTTP File server directory is " + baseDir)
    httpServer = new HttpServer(conf, baseDir, securityManager, requestedPort, "HTTP file server")
    httpServer.start()
    serverUri = httpServer.uri
    logDebug("HTTP file server started at: " + serverUri)
}

            /**
             * Runs a timer task to periodically定期地 clean up metadata (e.g. old files or hashtable entries)
             */
            private[spark] class MetadataCleaner(
                cleanerType: MetadataCleanerType.MetadataCleanerType,
                cleanupFunc: (Long) => Unit,
                conf: SparkConf)
              extends Logging
            {
              val name = cleanerType.toString

              private val delaySeconds = MetadataCleaner.getDelaySeconds(conf, cleanerType)
              private val periodSeconds = math.max(10, delaySeconds / 10)
              private val timer = new Timer(name + " cleanup timer", true)


              private val task = new TimerTask {
                override def run() {
                  try {
                  //就这，定期清理，用cleanupFunc
                    cleanupFunc(System.currentTimeMillis() - (delaySeconds * 1000))
                    logInfo("Ran metadata cleaner for " + name)
                  } catch {
                    case e: Exception => logError("Error running cleanup task for " + name, e)
                  }
                }
              }

              if (delaySeconds > 0) {
                logDebug(
                  "Starting metadata cleaner for " + name + " with delay of " + delaySeconds + " seconds " +
                  "and period of " + periodSeconds + " secs")
                timer.schedule(task, delaySeconds * 1000, periodSeconds * 1000)
              }

              def cancel() {
                timer.cancel()
              }
            }
而clean函数如下

            private[spark] def cleanup(cleanupTime:Long){                       persistentRdds.clearOldValues(cleanupTime)
            }

private val EVENT_QUEUE_CAPACITY = 10000
  private val eventQueue = new LinkedBlockingQueue[SparkListenerEvent](EVENT_QUEUE_CAPACITY)
  private var queueFullErrorMessageLogged = false
  private var started = false
  // A counter that represents the number of events produced and consumed in the queue
  private val eventLock = new Semaphore(0)

  private val listenerThread = new Thread("SparkListenerBus") {
    setDaemon(true)
    override def run(): Unit = Utils.logUncaughtExceptions {
      while (true) {
        eventLock.acquire()
        // Atomically remove and process this event
        LiveListenerBus.this.synchronized {
          val event = eventQueue.poll
          if (event == SparkListenerShutdown) {
            // Get out of the while loop and shutdown the daemon thread
            return
          }
          Option(event).foreach(postToAll)
        }
      }
    }
  }

  def start() {
    if (started) {
      throw new IllegalStateException("Listener bus already started!")
    }
    listenerThread.start()
    started = true
  }
def post(event: SparkListenerEvent) {
    val eventAdded = eventQueue.offer(event)
    if (eventAdded) {
      eventLock.release()
    } else {
      logQueueFullErrorMessage()
    }
  }

  def listenerThreadIsAlive: Boolean = synchronized { listenerThread.isAlive }

  def queueIsEmpty: Boolean = synchronized { eventQueue.isEmpty }

  def stop() {
    if (!started) {
      throw new IllegalStateException("Attempted to stop a listener bus that has not yet started!")
    }
    post(SparkListenerShutdown)
    listenerThread.join()
  }

  private def registerSources() {
    val instConfig = metricsConfig.getInstance(instance)
    val sourceConfigs = metricsConfig.subProperties(instConfig, MetricsSystem.SOURCE_REGEX)

    // Register all the sources related to instance
    sourceConfigs.foreach { kv =>
      val classPath = kv._2.getProperty("class")
      try {
        val source = Utils.classForName(classPath).newInstance()
        registerSource(source.asInstanceOf[Source])
      } catch {
        case e: Exception => logError("Source class " + classPath + " cannot be instantiated", e)
      }
    }
  }

  def start(): Unit = {
    listenerBus.addListener(listener)

    val scheduleTask = new Runnable() {
      override def run(): Unit = {
        try {
          schedule()
        } catch {
          case ct: ControlThrowable =>
            throw ct
          case t: Throwable =>
            logWarning(s"Uncaught exception in thread ${Thread.currentThread().getName}", t)
        }
      }
    }
    executor.scheduleAtFixedRate(scheduleTask, 0, intervalMillis, TimeUnit.MILLISECONDS)
  }

/** Keep cleaning RDD, shuffle, and broadcast state. */
  private def keepCleaning(): Unit = Utils.tryOrStopSparkContext(sc) {
    while (!stopped) {
      try {
        val reference = Option(referenceQueue.remove(ContextCleaner.REF_QUEUE_POLL_TIMEOUT))
          .map(_.asInstanceOf[CleanupTaskWeakReference])

// Synchronize here to avoid being interrupted on stop()
        synchronized {
          reference.map(_.task).foreach { task =>
            logDebug("Got cleaning task " + task)
            referenceBuffer -= reference.get
            task match {
              case CleanRDD(rddId) =>
                doCleanupRDD(rddId, blocking = blockOnCleanupTasks)
              case CleanShuffle(shuffleId) =>
                doCleanupShuffle(shuffleId, blocking = blockOnShuffleCleanupTasks)
              case CleanBroadcast(broadcastId) =>
                doCleanupBroadcast(broadcastId, blocking = blockOnCleanupTasks)
              case CleanAccum(accId) =>
                doCleanupAccum(accId, blocking = blockOnCleanupTasks)
              case CleanCheckpoint(rddId) =>
                doCleanCheckpoint(rddId)
            }
          }
        }
      } catch {
        case ie: InterruptedException if stopped => // ignore
        case e: Exception => logError("Error in cleaning thread", e)
      }
    }
  }

private def initDriveMetrics(){
 SparkEnv.get.metricsSystem.registerSource(dagSchedulerSource)
 SparkEnv.get.metricsSystem.registerSource(blockManagerSource)
}
initDriveMetrics()