12345678910111213141516171819202122232425262728293031

private[spark] trait RpcEndpoint {

  /**   * The [[RpcEnv]] that this [[RpcEndpoint]] is registered to.   */  val rpcEnv: RpcEnv   ....}

 /**   * Process messages from [[RpcEndpointRef.send]] or [[RpcCallContext.reply)]]. If receiving a   * unmatched message, [[SparkException]] will be thrown and sent to `onError`.   */  def receive: PartialFunction[Any, Unit] = {    case _ => throw new SparkException(self + " does not implement 'receive'")  }

 /**   * Process messages from [[RpcEndpointRef.ask]]. If receiving a unmatched message,   * [[SparkException]] will be thrown and sent to `onError`.   */  def receiveAndReply(context: RpcCallContext): PartialFunction[Any, Unit] = {    case _ => context.sendFailure(new SparkException(self + " won't reply anything"))  }

  /**   * Invoked before [[RpcEndpoint]] starts to handle any message.   */  def onStart(): Unit = {    // By default, do nothing.  }

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192

private[netty] class NettyRpcEnv(    val conf: SparkConf,    javaSerializerInstance: JavaSerializerInstance,    host: String,    securityManager: SecurityManager) extends RpcEnv(conf) with Logging {

  private[netty] val transportConf = SparkTransportConf.fromSparkConf(    conf.clone.set("spark.rpc.io.numConnectionsPerPeer", "1"),    "rpc",    conf.getInt("spark.rpc.io.threads", 0))

  // 设置一个消息分发器  private val dispatcher: Dispatcher = new Dispatcher(this)

  private val streamManager = new NettyStreamManager(this)

  private val transportContext = new TransportContext(transportConf,    new NettyRpcHandler(dispatcher, this, streamManager))

  private def createClientBootstraps(): java.util.List[TransportClientBootstrap] = {    if (securityManager.isAuthenticationEnabled()) {      java.util.Arrays.asList(new SaslClientBootstrap(transportConf, "", securityManager,        securityManager.isSaslEncryptionEnabled()))    } else {      java.util.Collections.emptyList[TransportClientBootstrap]    }  }

  private val clientFactory = transportContext.createClientFactory(createClientBootstraps())

  /**   * A separate client factory for file downloads. This avoids using the same RPC handler as   * the main RPC context, so that events caused by these clients are kept isolated from the   * main RPC traffic.   *   * It also allows for different configuration of certain properties, such as the number of   * connections per peer.   */  @volatile private var fileDownloadFactory: TransportClientFactory = _

  val timeoutScheduler = ThreadUtils.newDaemonSingleThreadScheduledExecutor("netty-rpc-env-timeout")

  // Because TransportClientFactory.createClient is blocking, we need to run it in this thread pool  // to implement non-blocking send/ask.  // TODO: a non-blocking TransportClientFactory.createClient in future  private[netty] val clientConnectionExecutor = ThreadUtils.newDaemonCachedThreadPool(    "netty-rpc-connection",    conf.getInt("spark.rpc.connect.threads", 64))

  @volatile private var server: TransportServer = _

  private val stopped = new AtomicBoolean(false)

  /**   * A map for [[RpcAddress]] and [[Outbox]]. When we are connecting to a remote [[RpcAddress]],   * we just put messages to its [[Outbox]] to implement a non-blocking `send` method.   */    // 多个地址对应的发件箱  private val outboxes = new ConcurrentHashMap[RpcAddress, Outbox]()

  /**   * Remove the address's Outbox and stop it.   */  private[netty] def removeOutbox(address: RpcAddress): Unit = {    val outbox = outboxes.remove(address)    if (outbox != null) {      outbox.stop()    }  }  // 启动TransportServer来接收远程消息  def startServer(bindAddress: String, port: Int): Unit = {    val bootstraps: java.util.List[TransportServerBootstrap] =      if (securityManager.isAuthenticationEnabled()) {        java.util.Arrays.asList(new SaslServerBootstrap(transportConf, securityManager))      } else {        java.util.Collections.emptyList()      }    server = transportContext.createServer(bindAddress, port, bootstraps)    dispatcher.registerRpcEndpoint(      RpcEndpointVerifier.NAME, new RpcEndpointVerifier(this, dispatcher))  }

  @Nullable  override lazy val address: RpcAddress = {    if (server != null) RpcAddress(host, server.getPort()) else null  }

  // 注册当前端点  override def setupEndpoint(name: String, endpoint: RpcEndpoint): RpcEndpointRef = {    dispatcher.registerRpcEndpoint(name, endpoint)  }....

123456

private class EndpointData(    val name: String,    val endpoint: RpcEndpoint,    val ref: NettyRpcEndpointRef) {  val inbox = new Inbox(ref, endpoint)}

12345678910111213141516

# Start all spark daemons.# Starts the master on this node.# Starts a worker on each node specified in conf/slaves

if [ -z "${SPARK_HOME}" ]; then  #如果没有发现Spark环境变量  export SPARK_HOME="$(cd "`dirname "$0"`"/..; pwd)" # 获得当前的目录把当前目录设置为SPARK_HOMEfi

# Load the Spark configuration. "${SPARK_HOME}/sbin/spark-config.sh" #加载 spark-config.sh配置

# Start Master"${SPARK_HOME}/sbin"/start-master.sh

# Start Workers"${SPARK_HOME}/sbin"/start-slaves.sh

1234567891011121314151617

# included in all the spark scripts with source command# should not be executable directly# also should not be passed any arguments, since we need original $*

# symlink and absolute path should rely on SPARK_HOME to resolveif [ -z "${SPARK_HOME}" ]; then  export SPARK_HOME="$(cd "`dirname "$0"`"/..; pwd)"fi

export SPARK_CONF_DIR="${SPARK_CONF_DIR:-"${SPARK_HOME}/conf"}" #设置 SPARK_CONF_DIR 目录# Add the PySpark classes to the PYTHONPATH:if [ -z "${PYSPARK_PYTHONPATH_SET}" ]; then  export PYTHONPATH="${SPARK_HOME}/python:${PYTHONPATH}"  export PYTHONPATH="${SPARK_HOME}/python/lib/py4j-0.10.4-src.zip:${PYTHONPATH}"  export PYSPARK_PYTHONPATH_SET=1fiexport JAVA_HOME=/opt/module/jdk1.8.0_162

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455

# Unless required by applicable law or agreed to in writing, software# distributed under the License is distributed on an "AS IS" BASIS,# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.# See the License for the specific language governing permissions and# limitations under the License.#

# Starts the master on the machine this script is executed on.

if [ -z "${SPARK_HOME}" ]; then  export SPARK_HOME="$(cd "`dirname "$0"`"/..; pwd)"fi

# NOTE: This exact class name is matched downstream by SparkSubmit.# Any changes need to be reflected there.CLASS="org.apache.spark.deploy.master.Master"  #调用Master

if [[ "$@" = *--help ]] || [[ "$@" = *-h ]]; then  echo "Usage: ./sbin/start-master.sh [options]"  pattern="Usage:"  pattern+="\|Using Spark's default log4j profile:"  pattern+="\|Registered signal handlers for"

  "${SPARK_HOME}"/bin/spark-class $CLASS --help 2>&1 | grep -v "$pattern" 1>&2  exit 1fi

ORIGINAL_ARGS="$@"

. "${SPARK_HOME}/sbin/spark-config.sh"  

. "${SPARK_HOME}/bin/load-spark-env.sh" #加载环境变量

if [ "$SPARK_MASTER_PORT" = "" ]; then # 如果没有端口 默认7077  SPARK_MASTER_PORT=7077fi

if [ "$SPARK_MASTER_HOST" = "" ]; then  case `uname` in      (SunOS)           # 如果没有设置HOST 则把/usr/sbin/check-hostname作为主机名          SPARK_MASTER_HOST="`/usr/sbin/check-hostname | awk '{print $NF}'`"          ;;      (*)          SPARK_MASTER_HOST="`hostname -f`"          ;;  esacfi

if [ "$SPARK_MASTER_WEBUI_PORT" = "" ]; then  SPARK_MASTER_WEBUI_PORT=8080fi

"${SPARK_HOME}/sbin"/spark-daemon.sh start $CLASS 1 \  --host $SPARK_MASTER_HOST --port $SPARK_MASTER_PORT --webui-port $SPARK_MASTER_WEBUI_PORT \  $ORIGINAL_ARGS

1234567891011

def main(argStrings: Array[String]) {  // 1、初始化log对象  Utils.initDaemon(log)  // 2、加载SparkConf  val conf = new SparkConf  // 3、解析Master启动参数  val args = new MasterArguments(argStrings, conf)  // 4、启动RPC框架端点  val (rpcEnv, _, _) = startRpcEnvAndEndpoint(args.host, args.port, args.webUiPort, conf)  rpcEnv.awaitTermination()}

123456789101112131415161718192021222324252627

# Starts a slave instance on each machine specified in the conf/slaves file.

if [ -z "${SPARK_HOME}" ]; then  export SPARK_HOME="$(cd "`dirname "$0"`"/..; pwd)" #获取当前的目录fi

. "${SPARK_HOME}/sbin/spark-config.sh". "${SPARK_HOME}/bin/load-spark-env.sh" #加载配置

# Find the port number for the masterif [ "$SPARK_MASTER_PORT" = "" ]; then  SPARK_MASTER_PORT=7077fi

if [ "$SPARK_MASTER_HOST" = "" ]; then  case `uname` in      (SunOS)          SPARK_MASTER_HOST="`/usr/sbin/check-hostname | awk '{print $NF}'`"          ;;      (*)          SPARK_MASTER_HOST="`hostname -f`"          ;;  esacfi

# Launch the slaves 调用了start-slave.sh"${SPARK_HOME}/sbin/slaves.sh" cd "${SPARK_HOME}" \; "${SPARK_HOME}/sbin/start-slave.sh" "spark://$SPARK_MASTER_HOST:$SPARK_MASTER_PORT"

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172

# Starts a slave on the machine this script is executed on.## Environment Variables##   SPARK_WORKER_INSTANCES  The number of worker instances to run on this#                           slave.  Default is 1.#   SPARK_WORKER_PORT       The base port number for the first worker. If set,#                           subsequent workers will increment this number.  If#                           unset, Spark will find a valid port number, but#                           with no guarantee of a predictable pattern.#   SPARK_WORKER_WEBUI_PORT The base port for the web interface of the first#                           worker.  Subsequent workers will increment this#                           number.  Default is 8081.

if [ -z "${SPARK_HOME}" ]; then  export SPARK_HOME="$(cd "`dirname "$0"`"/..; pwd)"fi

# NOTE: This exact class name is matched downstream by SparkSubmit.# Any changes need to be reflected there.CLASS="org.apache.spark.deploy.worker.Worker"

if [[ $# -lt 1 ]] || [[ "$@" = *--help ]] || [[ "$@" = *-h ]]; then  echo "Usage: ./sbin/start-slave.sh [options] <master>"  pattern="Usage:"  pattern+="\|Using Spark's default log4j profile:"  pattern+="\|Registered signal handlers for"

  "${SPARK_HOME}"/bin/spark-class $CLASS --help 2>&1 | grep -v "$pattern" 1>&2  exit 1fi

. "${SPARK_HOME}/sbin/spark-config.sh"

. "${SPARK_HOME}/bin/load-spark-env.sh"

# First argument should be the master; we need to store it aside because we may# need to insert arguments between it and the other argumentsMASTER=$1shift

# Determine desired worker portif [ "$SPARK_WORKER_WEBUI_PORT" = "" ]; then  SPARK_WORKER_WEBUI_PORT=8081fi

# Start up the appropriate number of workers on this machine.# quick local function to start a workerfunction start_instance {  WORKER_NUM=$1  shift

  if [ "$SPARK_WORKER_PORT" = "" ]; then    PORT_FLAG=    PORT_NUM=  else    PORT_FLAG="--port"    PORT_NUM=$(( $SPARK_WORKER_PORT + $WORKER_NUM - 1 ))  fi  WEBUI_PORT=$(( $SPARK_WORKER_WEBUI_PORT + $WORKER_NUM - 1 ))   #调用org.apache.spark.deploy.worker.Worker  "${SPARK_HOME}/sbin"/spark-daemon.sh start $CLASS $WORKER_NUM \     --webui-port "$WEBUI_PORT" $PORT_FLAG $PORT_NUM $MASTER "$@" }

if [ "$SPARK_WORKER_INSTANCES" = "" ]; then  start_instance 1 "$@"else  for ((i=0; i<$SPARK_WORKER_INSTANCES; i++)); do    start_instance $(( 1 + $i )) "$@"  donefi

12345678

def main(argStrings: Array[String]) {    Utils.initDaemon(log)    val conf = new SparkConf    val args = new WorkerArguments(argStrings, conf)    val rpcEnv = startRpcEnvAndEndpoint(args.host, args.port, args.webUiPort, args.cores,      args.memory, args.masters, args.workDir, conf = conf)    rpcEnv.awaitTermination()  }

12345678

if [ -z "${SPARK_HOME}" ]; then  source "$(dirname "$0")"/find-spark-homefi

# disable randomized hash for string in Python 3.3+export PYTHONHASHSEED=0

exec "${SPARK_HOME}"/bin/spark-class org.apache.spark.deploy.SparkSubmit "$@"

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778

if [ -z "${SPARK_HOME}" ]; then  source "$(dirname "$0")"/find-spark-homefi

. "${SPARK_HOME}"/bin/load-spark-env.sh

# Find the java binaryif [ -n "${JAVA_HOME}" ]; then  RUNNER="${JAVA_HOME}/bin/java"else  if [ "$(command -v java)" ]; then    RUNNER="java"  else    echo "JAVA_HOME is not set" >&2    exit 1  fifi

# Find Spark jars.if [ -d "${SPARK_HOME}/jars" ]; then  SPARK_JARS_DIR="${SPARK_HOME}/jars"else  SPARK_JARS_DIR="${SPARK_HOME}/assembly/target/scala-$SPARK_SCALA_VERSION/jars"fi

if [ ! -d "$SPARK_JARS_DIR" ] && [ -z "$SPARK_TESTING$SPARK_SQL_TESTING" ]; then  echo "Failed to find Spark jars directory ($SPARK_JARS_DIR)." 1>&2  echo "You need to build Spark with the target \"package\" before running this program." 1>&2  exit 1else  LAUNCH_CLASSPATH="$SPARK_JARS_DIR/*"fi

# Add the launcher build dir to the classpath if requested.if [ -n "$SPARK_PREPEND_CLASSES" ]; then  LAUNCH_CLASSPATH="${SPARK_HOME}/launcher/target/scala-$SPARK_SCALA_VERSION/classes:$LAUNCH_CLASSPATH"fi

# For testsif [[ -n "$SPARK_TESTING" ]]; then  unset YARN_CONF_DIR  unset HADOOP_CONF_DIRfi

# The launcher library will print arguments separated by a NULL character, to allow arguments with# characters that would be otherwise interpreted by the shell. Read that in a while loop, populating# an array that will be used to exec the final command.## The exit code of the launcher is appended to the output, so the parent shell removes it from the# command array and checks the value to see if the launcher succeeded.build_command() {  "$RUNNER" -Xmx128m -cp "$LAUNCH_CLASSPATH" org.apache.spark.launcher.Main "$@"  printf "%d\0" $?}

CMD=()while IFS= read -d '' -r ARG; do  CMD+=("$ARG")done < <(build_command "$@")

COUNT=${#CMD[@]}LAST=$((COUNT - 1))LAUNCHER_EXIT_CODE=${CMD[$LAST]}

# Certain JVM failures result in errors being printed to stdout (instead of stderr), which causes# the code that parses the output of the launcher to get confused. In those cases, check if the# exit code is an integer, and if it's not, handle it as a special error case.if ! [[ $LAUNCHER_EXIT_CODE =~ ^[0-9]+$ ]]; then  echo "${CMD[@]}" | head -n-1 1>&2  exit 1fi

if [ $LAUNCHER_EXIT_CODE != 0 ]; then  exit $LAUNCHER_EXIT_CODEfi

CMD=("${CMD[@]:0:$LAST}")exec "${CMD[@]}"

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465

object SparkSubmit {

  // Cluster managers  private val YARN = 1  private val STANDALONE = 2  private val MESOS = 4  private val LOCAL = 8  private val ALL_CLUSTER_MGRS = YARN | STANDALONE | MESOS | LOCAL

  // Deploy modes  private val CLIENT = 1  private val CLUSTER = 2  private val ALL_DEPLOY_MODES = CLIENT | CLUSTER

  // Special primary resource names that represent shells rather than application jars.  private val SPARK_SHELL = "spark-shell"  private val PYSPARK_SHELL = "pyspark-shell"  private val SPARKR_SHELL = "sparkr-shell"  private val SPARKR_PACKAGE_ARCHIVE = "sparkr.zip"  private val R_PACKAGE_ARCHIVE = "rpkg.zip"

  private val CLASS_NOT_FOUND_EXIT_STATUS = 101

  // scalastyle:off println  // Exposed for testing  private[spark] var exitFn: Int => Unit = (exitCode: Int) => System.exit(exitCode)  private[spark] var printStream: PrintStream = System.err  private[spark] def printWarning(str: String): Unit = printStream.println("Warning: " + str)  private[spark] def printErrorAndExit(str: String): Unit = {    printStream.println("Error: " + str)    printStream.println("Run with --help for usage help or --verbose for debug output")    exitFn(1)  }  private[spark] def printVersionAndExit(): Unit = {    printStream.println("""Welcome to      ____              __     / __/__  ___ _____/ /__    _\ \/ _ \/ _ `/ __/  '_/   /___/ .__/\_,_/_/ /_/\_\   version %s      /_/                        """.format(SPARK_VERSION))      printStream.println("Using Scala %s, %s, %s".format(      Properties.versionString, Properties.javaVmName, Properties.javaVersion))    printStream.println("Branch %s".format(SPARK_BRANCH))    printStream.println("Compiled by user %s on %s".format(SPARK_BUILD_USER, SPARK_BUILD_DATE))    printStream.println("Revision %s".format(SPARK_REVISION))    printStream.println("Url %s".format(SPARK_REPO_URL))    printStream.println("Type --help for more information.")    exitFn(0)  }  // scalastyle:on println

  def main(args: Array[String]): Unit = {    val appArgs = new SparkSubmitArguments(args)    if (appArgs.verbose) {      // scalastyle:off println      printStream.println(appArgs)      // scalastyle:on println    }    appArgs.action match {      case SparkSubmitAction.SUBMIT => submit(appArgs)      case SparkSubmitAction.KILL => kill(appArgs)      case SparkSubmitAction.REQUEST_STATUS => requestStatus(appArgs)    }  }

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374

# Shell script for starting the Spark Shell REPL

cygwin=falsecase "$(uname)" in  CYGWIN*) cygwin=true;;esac

# Enter posix mode for bashset -o posix

if [ -z "${SPARK_HOME}" ]; then  source "$(dirname "$0")"/find-spark-homefi

export _SPARK_CMD_USAGE="Usage: ./bin/spark-shell [options]"

# SPARK-4161: scala does not assume use of the java classpath,# so we need to add the "-Dscala.usejavacp=true" flag manually. We# do this specifically for the Spark shell because the scala REPL# has its own class loader, and any additional classpath specified# through spark.driver.extraClassPath is not automatically propagated.SPARK_SUBMIT_OPTS="$SPARK_SUBMIT_OPTS -Dscala.usejavacp=true"

function main() {  if $cygwin; then    # Workaround for issue involving JLine and Cygwin    # (see http://sourceforge.net/p/jline/bugs/40/).    # If you're using the Mintty terminal emulator in Cygwin, may need to set the    # "Backspace sends ^H" setting in "Keys" section of the Mintty options    # (see https://github.com/sbt/sbt/issues/562).    stty -icanon min 1 -echo > /dev/null 2>&1    export SPARK_SUBMIT_OPTS="$SPARK_SUBMIT_OPTS -Djline.terminal=unix"    "${SPARK_HOME}"/bin/spark-submit --class org.apache.spark.repl.Main --name "Spark shell" "$@"    stty icanon echo > /dev/null 2>&1  else    export SPARK_SUBMIT_OPTS    "${SPARK_HOME}"/bin/spark-submit --class org.apache.spark.repl.Main --name "Spark shell" "$@"  fi  }

# Copy restore-TTY-on-exit functions from Scala script so spark-shell exits properly even in# binary distribution of Spark where Scala is not installedexit_status=127saved_stty=""

# restore stty settings (echo in particular)function restoreSttySettings() {  stty $saved_stty  saved_stty=""}

function onExit() {  if [[ "$saved_stty" != "" ]]; then    restoreSttySettings  fi  exit $exit_status}

# to reenable echo if we are interrupted before completing.trap onExit INT

# save terminal settingssaved_stty=$(stty -g 2>/dev/null)# clear on error so we don't later try to restore themif [[ ! $? ]]; then  saved_stty=""fi

main "$@" #调用的main函数 最终执行的依旧是spark-submi

# record the exit status lest it be overwritten:# then reenable echo and propagate the code.exit_status=$?onExit

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788

package org.apache.spark.repl

object Main extends Logging {

  initializeLogIfNecessary(true)  Signaling.cancelOnInterrupt()

  val conf = new SparkConf()  val rootDir = conf.getOption("spark.repl.classdir").getOrElse(Utils.getLocalDir(conf))  val outputDir = Utils.createTempDir(root = rootDir, namePrefix = "repl")

  var sparkContext: SparkContext = _  var sparkSession: SparkSession = _  // this is a public var because tests reset it.  var interp: SparkILoop = _

  private var hasErrors = false

  private def scalaOptionError(msg: String): Unit = {    hasErrors = true    Console.err.println(msg)  }

  def main(args: Array[String]) {    doMain(args, new SparkILoop)  }

  // Visible for testing  private[repl] def doMain(args: Array[String], _interp: SparkILoop): Unit = {    interp = _interp    val jars = Utils.getUserJars(conf, isShell = true).mkString(File.pathSeparator)    val interpArguments = List(      "-Yrepl-class-based",      "-Yrepl-outdir", s"${outputDir.getAbsolutePath}",      "-classpath", jars    ) ++ args.toList

    val settings = new GenericRunnerSettings(scalaOptionError)    settings.processArguments(interpArguments, true)

    if (!hasErrors) {      interp.process(settings) // Repl starts and goes in loop of R.E.P.L      Option(sparkContext).map(_.stop)    }  }

  def createSparkSession(): SparkSession = {    val execUri = System.getenv("SPARK_EXECUTOR_URI")    conf.setIfMissing("spark.app.name", "Spark shell")    // SparkContext will detect this configuration and register it with the RpcEnv's    // file server, setting spark.repl.class.uri to the actual URI for executors to    // use. This is sort of ugly but since executors are started as part of SparkContext    // initialization in certain cases, there's an initialization order issue that prevents    // this from being set after SparkContext is instantiated.    conf.set("spark.repl.class.outputDir", outputDir.getAbsolutePath())    if (execUri != null) {      conf.set("spark.executor.uri", execUri)    }    if (System.getenv("SPARK_HOME") != null) {      conf.setSparkHome(System.getenv("SPARK_HOME"))    }

    val builder = SparkSession.builder.config(conf)    if (conf.get(CATALOG_IMPLEMENTATION.key, "hive").toLowerCase == "hive") {      if (SparkSession.hiveClassesArePresent) {        // In the case that the property is not set at all, builder's config        // does not have this value set to 'hive' yet. The original default        // behavior is that when there are hive classes, we use hive catalog.        sparkSession = builder.enableHiveSupport().getOrCreate()        logInfo("Created Spark session with Hive support")      } else {        // Need to change it back to 'in-memory' if no hive classes are found        // in the case that the property is set to hive in spark-defaults.conf        builder.config(CATALOG_IMPLEMENTATION.key, "in-memory")        sparkSession = builder.getOrCreate()        logInfo("Created Spark session")      }    } else {      // In the case that the property is set but not to 'hive', the internal      // default is 'in-memory'. So the sparkSession will use in-memory catalog.      sparkSession = builder.getOrCreate()      logInfo("Created Spark session")    }    sparkContext = sparkSession.sparkContext    sparkSession  }

}

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149

private[repl] trait SparkILoopInit {  self: SparkILoop =>

  /** Print a welcome message */  def printWelcome() {    echo("""Welcome to      ____              __     / __/__  ___ _____/ /__    _\ \/ _ \/ _ `/ __/  '_/   /___/ .__/\_,_/_/ /_/\_\   version %s      /_/""".format(SPARK_VERSION))    import Properties._    val welcomeMsg = "Using Scala %s (%s, Java %s)".format(      versionString, javaVmName, javaVersion)    echo(welcomeMsg)    echo("Type in expressions to have them evaluated.")    echo("Type :help for more information.")   }

  protected def asyncMessage(msg: String) {    if (isReplInfo || isReplPower)      echoAndRefresh(msg)  }

  private val initLock = new java.util.concurrent.locks.ReentrantLock()  private val initCompilerCondition = initLock.newCondition() // signal the compiler is initialized  private val initLoopCondition = initLock.newCondition()     // signal the whole repl is initialized  private val initStart = System.nanoTime

  private def withLock[T](body: => T): T = {    initLock.lock()    try body    finally initLock.unlock()  }  // a condition used to ensure serial access to the compiler.  @volatile private var initIsComplete = false  @volatile private var initError: String = null  private def elapsed() = "%.3f".format((System.nanoTime - initStart).toDouble / 1000000000L)

  // the method to be called when the interpreter is initialized.  // Very important this method does nothing synchronous (i.e. do  // not try to use the interpreter) because until it returns, the  // repl's lazy val `global` is still locked.  protected def initializedCallback() = withLock(initCompilerCondition.signal())

  // Spins off a thread which awaits a single message once the interpreter  // has been initialized.  protected def createAsyncListener() = {    io.spawn {      withLock(initCompilerCondition.await())      asyncMessage("[info] compiler init time: " + elapsed() + " s.")      postInitialization()    }  }

  // called from main repl loop  protected def awaitInitialized(): Boolean = {    if (!initIsComplete)      withLock { while (!initIsComplete) initLoopCondition.await() }    if (initError != null) {      // scalastyle:off println      println("""        |Failed to initialize the REPL due to an unexpected error.        |This is a bug, please, report it along with the error diagnostics printed below.        |%s.""".stripMargin.format(initError)      )      // scalastyle:on println      false    } else true  }  // private def warningsThunks = List(  //   () => intp.bind("lastWarnings", "" + typeTag[List[(Position, String)]], intp.lastWarnings _),  // )

  protected def postInitThunks = List[Option[() => Unit]](    Some(intp.setContextClassLoader _),    if (isReplPower) Some(() => enablePowerMode(true)) else None  ).flatten  // ++ (  //   warningsThunks  // )  // called once after init condition is signalled  protected def postInitialization() {    try {      postInitThunks foreach (f => addThunk(f()))      runThunks()    } catch {      case ex: Throwable =>        initError = stackTraceString(ex)        throw ex    } finally {      initIsComplete = true

      if (isAsync) {        asyncMessage("[info] total init time: " + elapsed() + " s.")        withLock(initLoopCondition.signal())      }    }  }

  def initializeSpark() {    intp.beQuietDuring {      command("""        @transient val spark = org.apache.spark.repl.Main.interp.createSparkSession()        @transient val sc = {          val _sc = spark.sparkContext          if (_sc.getConf.getBoolean("spark.ui.reverseProxy", false)) {            val proxyUrl = _sc.getConf.get("spark.ui.reverseProxyUrl", null)            if (proxyUrl != null) {              println(s"Spark Context Web UI is available at ${proxyUrl}/proxy/${_sc.applicationId}")            } else {              println(s"Spark Context Web UI is available at Spark Master Public URL")            }          } else {            _sc.uiWebUrl.foreach {              webUrl => println(s"Spark context Web UI available at ${webUrl}")            }          }          println("Spark context available as 'sc' " +            s"(master = ${_sc.master}, app id = ${_sc.applicationId}).")          println("Spark session available as 'spark'.")          _sc        }        """)      command("import org.apache.spark.SparkContext._")      command("import spark.implicits._")      command("import spark.sql")      command("import org.apache.spark.sql.functions._")    }  }

  // code to be executed only after the interpreter is initialized  // and the lazy val `global` can be accessed without risk of deadlock.  private var pendingThunks: List[() => Unit] = Nil  protected def addThunk(body: => Unit) = synchronized {    pendingThunks :+= (() => body)  }  protected def runThunks(): Unit = synchronized {    if (pendingThunks.nonEmpty)      logDebug("Clearing " + pendingThunks.size + " thunks.")

    while (pendingThunks.nonEmpty) {      val thunk = pendingThunks.head      pendingThunks = pendingThunks.tail      thunk()    }  }}