SparkContext — Entry Point to Spark Core

SparkContext (aka Spark context) is the entry point to Spark’s core services for a Spark application.

Note
You could also assume that a SparkContext instance is a Spark application.

Spark context sets up internal services and establishes a connection to a Spark execution environment.

Once a SparkContext is created you can use it to create RDDs, accumulators and broadcast variables, access Spark services and run jobs (until SparkContext is stopped).

A Spark context is essentially a client of Spark’s execution environment and acts as the master of your Spark application (don’t get confused with the other meaning of Master in Spark, though).

sparkcontext services.png
Figure 1. Spark context acts as the master of your Spark application

SparkContext offers the following functions:

Table 1. SparkContext’s Internal Registries and Counters
Name Description

persistentRdds

Lookup table of persistent/cached RDDs per their ids.

Used when SparkContext is requested to:

Table 2. SparkContext’s Internal Properties
Name Initial Value Description

_taskScheduler

(uninitialized)

TaskScheduler

Tip
Read the scaladoc of org.apache.spark.SparkContext.
Tip

Enable INFO logging level for org.apache.spark.SparkContext logger to see what happens inside.

Add the following line to conf/log4j.properties:

log4j.logger.org.apache.spark.SparkContext=INFO

Refer to Logging.

getRDDStorageInfo Method

Caution
FIXME

unpersistRDD Internal Method

unpersistRDD(rddId: Int, blocking: Boolean = true): Unit
Caution
FIXME

Accessing persistent RDDs — getPersistentRDDs Method

getPersistentRDDs: Map[Int, RDD[_]]

getPersistentRDDs returns the collection of RDDs that have marked themselves as persistent via cache.

Internally, getPersistentRDDs returns persistentRdds internal registry.

Cancelling Job — cancelJob Method

cancelJob(jobId: Int)

cancelJob requests DAGScheduler to cancel a Spark job jobId.

Programmable Dynamic Allocation

SparkContext offers the following methods as the developer API for dynamic allocation of executors:

Requesting New Executors — requestExecutors Method

requestExecutors(numAdditionalExecutors: Int): Boolean

requestExecutors requests numAdditionalExecutors executors from CoarseGrainedSchedulerBackend.

Requesting to Kill Executors — killExecutors Method

killExecutors(executorIds: Seq[String]): Boolean
Caution
FIXME

Requesting Total Executors — requestTotalExecutors Method

requestTotalExecutors(
  numExecutors: Int,
  localityAwareTasks: Int,
  hostToLocalTaskCount: Map[String, Int]): Boolean

requestTotalExecutors is a private[spark] method that requests the exact number of executors from a coarse-grained scheduler backend.

Note
It works for coarse-grained scheduler backends only.

When called for other scheduler backends you should see the following WARN message in the logs:

WARN Requesting executors is only supported in coarse-grained mode

Getting Executor Ids — getExecutorIds Method

Note
It works for coarse-grained scheduler backends only.

When called for other scheduler backends you should see the following WARN message in the logs:

WARN Requesting executors is only supported in coarse-grained mode
Caution
FIXME Why does SparkContext implement the method for coarse-grained scheduler backends? Why doesn’t SparkContext throw an exception when the method is called? Nobody seems to be using it (!)

Creating SparkContext Instance

You can create a SparkContext instance with or without creating a SparkConf object first.

Note
You may want to read Inside Creating SparkContext to learn what happens behind the scenes when SparkContext is created.

Getting Existing or Creating New SparkContext — getOrCreate Methods

getOrCreate(): SparkContext
getOrCreate(conf: SparkConf): SparkContext

getOrCreate methods allow you to get the existing SparkContext or create a new one.

import org.apache.spark.SparkContext
val sc = SparkContext.getOrCreate()

// Using an explicit SparkConf object
import org.apache.spark.SparkConf
val conf = new SparkConf()
  .setMaster("local[*]")
  .setAppName("SparkMe App")
val sc = SparkContext.getOrCreate(conf)

The no-param getOrCreate method requires that the two mandatory Spark settings - master and application name - are specified using spark-submit.

Constructors

SparkContext()
SparkContext(conf: SparkConf)
SparkContext(master: String, appName: String, conf: SparkConf)
SparkContext(
  master: String,
  appName: String,
  sparkHome: String = null,
  jars: Seq[String] = Nil,
  environment: Map[String, String] = Map())

You can create a SparkContext instance using the four constructors.

import org.apache.spark.SparkConf
val conf = new SparkConf()
  .setMaster("local[*]")
  .setAppName("SparkMe App")

import org.apache.spark.SparkContext
val sc = new SparkContext(conf)

When a Spark context starts up you should see the following INFO in the logs (amongst the other messages that come from the Spark services):

INFO SparkContext: Running Spark version 2.0.0-SNAPSHOT
Note
Only one SparkContext may be running in a single JVM (check out SPARK-2243 Support multiple SparkContexts in the same JVM). Sharing access to a SparkContext in the JVM is the solution to share data within Spark (without relying on other means of data sharing using external data stores).

Accessing Current SparkEnv — env Method

Caution
FIXME

Getting Current SparkConf — getConf Method

getConf: SparkConf

getConf returns the current SparkConf.

Note
Changing the SparkConf object does not change the current configuration (as the method returns a copy).

Getting Deployment Environment — master Method

master: String

master method returns the current value of spark.master which is the deployment environment in use.

Getting Application Name — appName Method

appName: String

appName returns the value of the mandatory spark.app.name setting.

Note
appName is used when SparkDeploySchedulerBackend starts, SparkUI creates a web UI, when postApplicationStart is executed, and for Mesos and checkpointing in Spark Streaming.

Getting Deploy Mode — deployMode Method

deployMode: String

deployMode returns the current value of spark.submit.deployMode setting or client if not set.

Getting Scheduling Mode — getSchedulingMode Method

getSchedulingMode: SchedulingMode.SchedulingMode

getSchedulingMode returns the current Scheduling Mode.

Getting Schedulable (Pool) by Name — getPoolForName Method

getPoolForName(pool: String): Option[Schedulable]

getPoolForName returns a Schedulable by the pool name, if one exists.

Note
getPoolForName is part of the Developer’s API and may change in the future.

It is exclusively used to show pool details in web UI (for a stage).

Getting All Pools — getAllPools Method

getAllPools: Seq[Schedulable]

getAllPools collects the Pools in TaskScheduler.rootPool.

Note
TaskScheduler.rootPool is part of the TaskScheduler Contract.
Note
getAllPools is part of the Developer’s API.
Caution
FIXME Where is the method used?
Note
getAllPools is used to calculate pool names for Stages tab in web UI with FAIR scheduling mode used.

Getting Default Level of Parallelism

defaultParallelism: Int

defaultParallelism requests TaskScheduler for the default level of parallelism.

Note
Default level of parallelism specifies the number of partitions in RDDs when created without specifying them explicitly by a user.
Note

defaultParallelism is used in SparkContext.parallelize, SparkContext.range and SparkContext.makeRDD (as well as Spark Streaming’s DStream.countByValue and DStream.countByValueAndWindow et al.).

defaultParallelism is also used to instantiate HashPartitioner and for the minimum number of partitions in HadoopRDDs.

taskScheduler Property

taskScheduler: TaskScheduler
taskScheduler_=(ts: TaskScheduler): Unit

taskScheduler manages (i.e. reads or writes) _taskScheduler internal property.

Getting Spark Version — version Property

version: String

version returns the Spark version this SparkContext uses.

makeRDD Method

Caution
FIXME

Submitting Jobs Asynchronously — submitJob Method

submitJob[T, U, R](
  rdd: RDD[T],
  processPartition: Iterator[T] => U,
  partitions: Seq[Int],
  resultHandler: (Int, U) => Unit,
  resultFunc: => R): SimpleFutureAction[R]

submitJob submits a job in an asynchronous, non-blocking way to DAGScheduler.

It cleans the processPartition input function argument and returns an instance of SimpleFutureAction that holds the JobWaiter instance.

Caution
FIXME What are resultFunc?

It is used in:

Spark Configuration

Caution
FIXME

SparkContext and RDDs

You use a Spark context to create RDDs (see Creating RDD).

When an RDD is created, it belongs to and is completely owned by the Spark context it originated from. RDDs can’t by design be shared between SparkContexts.

sparkcontext rdds.png
Figure 2. A Spark context creates a living space for RDDs.

Creating RDD — parallelize Method

SparkContext allows you to create many different RDDs from input sources like:

  • Scala’s collections, i.e. sc.parallelize(0 to 100)

  • local or remote filesystems, i.e. sc.textFile("README.md")

  • Any Hadoop InputSource using sc.newAPIHadoopFile

Unpersisting RDDs (Marking RDDs as non-persistent) — unpersist Method

It removes an RDD from the master’s Block Manager (calls removeRdd(rddId: Int, blocking: Boolean)) and the internal persistentRdds mapping.

It finally posts SparkListenerUnpersistRDD message to listenerBus.

Setting Checkpoint Directory — setCheckpointDir Method

setCheckpointDir(directory: String)

setCheckpointDir method is used to set up the checkpoint directory…​FIXME

Caution
FIXME

Registering Accumulator — register Methods

register(acc: AccumulatorV2[_, _]): Unit
register(acc: AccumulatorV2[_, _], name: String): Unit

register registers the acc accumulator. You can optionally give an accumulator a name.

Tip
You can create built-in accumulators for longs, doubles, and collection types using specialized methods.

Internally, register registers acc accumulator (with the current SparkContext).

Creating Built-In Accumulators

longAccumulator: LongAccumulator
longAccumulator(name: String): LongAccumulator
doubleAccumulator: DoubleAccumulator
doubleAccumulator(name: String): DoubleAccumulator
collectionAccumulator[T]: CollectionAccumulator[T]
collectionAccumulator[T](name: String): CollectionAccumulator[T]

You can use longAccumulator, doubleAccumulator or collectionAccumulator to create and register accumulators for simple and collection values.

longAccumulator returns LongAccumulator with the zero value 0.

doubleAccumulator returns DoubleAccumulator with the zero value 0.0.

collectionAccumulator returns CollectionAccumulator with the zero value java.util.List[T].

scala> val acc = sc.longAccumulator
acc: org.apache.spark.util.LongAccumulator = LongAccumulator(id: 0, name: None, value: 0)

scala> val counter = sc.longAccumulator("counter")
counter: org.apache.spark.util.LongAccumulator = LongAccumulator(id: 1, name: Some(counter), value: 0)

scala> counter.value
res0: Long = 0

scala> sc.parallelize(0 to 9).foreach(n => counter.add(n))

scala> counter.value
res3: Long = 45

The name input parameter allows you to give a name to an accumulator and have it displayed in Spark UI (under Stages tab for a given stage).

spark webui accumulators.png
Figure 3. Accumulators in the Spark UI
Tip
You can register custom accumulators using register methods.

Creating Broadcast Variable — broadcast Method

broadcast[T](value: T): Broadcast[T]

broadcast method creates a broadcast variable. It is a shared memory with value (as broadcast blocks) on the driver and later on all Spark executors.

val sc: SparkContext = ???
scala> val hello = sc.broadcast("hello")
hello: org.apache.spark.broadcast.Broadcast[String] = Broadcast(0)

Spark transfers the value to Spark executors once, and tasks can share it without incurring repetitive network transmissions when the broadcast variable is used multiple times.

sparkcontext broadcast executors.png
Figure 4. Broadcasting a value to executors
Note
The current BroadcastManager is available using SparkEnv.broadcastManager attribute and is always BroadcastManager (with few internal configuration changes to reflect where it runs, i.e. inside the driver or executors).

You should see the following INFO message in the logs:

INFO SparkContext: Created broadcast [id] from [callSite]

If ContextCleaner is defined, the new broadcast variable is registered for cleanup.

Note

Spark does not support broadcasting RDDs.

scala> sc.broadcast(sc.range(0, 10))
java.lang.IllegalArgumentException: requirement failed: Can not directly broadcast RDDs; instead, call collect() and broadcast the result.
  at scala.Predef$.require(Predef.scala:224)
  at org.apache.spark.SparkContext.broadcast(SparkContext.scala:1392)
  ... 48 elided

Once created, the broadcast variable (and other blocks) are displayed per executor and the driver in web UI (under Executors tab).

spark broadcast webui executors rdd blocks.png
Figure 5. Broadcast Variables In web UI’s Executors Tab

Distribute JARs to workers

The jar you specify with SparkContext.addJar will be copied to all the worker nodes.

The configuration setting spark.jars is a comma-separated list of jar paths to be included in all tasks executed from this SparkContext. A path can either be a local file, a file in HDFS (or other Hadoop-supported filesystems), an HTTP, HTTPS or FTP URI, or local:/path for a file on every worker node.

scala> sc.addJar("build.sbt")
15/11/11 21:54:54 INFO SparkContext: Added JAR build.sbt at http://192.168.1.4:49427/jars/build.sbt with timestamp 1447275294457
Caution
FIXME Why is HttpFileServer used for addJar?

SparkContext as Application-Wide Counter

SparkContext keeps track of:

Running Job Synchronously — runJob Methods

RDD actions run jobs using one of runJob methods.

runJob[T, U](
  rdd: RDD[T],
  func: (TaskContext, Iterator[T]) => U,
  partitions: Seq[Int],
  resultHandler: (Int, U) => Unit): Unit
runJob[T, U](
  rdd: RDD[T],
  func: (TaskContext, Iterator[T]) => U,
  partitions: Seq[Int]): Array[U]
runJob[T, U](
  rdd: RDD[T],
  func: Iterator[T] => U,
  partitions: Seq[Int]): Array[U]
runJob[T, U](rdd: RDD[T], func: (TaskContext, Iterator[T]) => U): Array[U]
runJob[T, U](rdd: RDD[T], func: Iterator[T] => U): Array[U]
runJob[T, U](
  rdd: RDD[T],
  processPartition: (TaskContext, Iterator[T]) => U,
  resultHandler: (Int, U) => Unit)
runJob[T, U: ClassTag](
  rdd: RDD[T],
  processPartition: Iterator[T] => U,
  resultHandler: (Int, U) => Unit)

runJob executes a function on one or many partitions of a RDD (in a SparkContext space) to produce a collection of values per partition.

Note
runJob can only work when a SparkContext is not stopped.

Internally, runJob first makes sure that the SparkContext is not stopped. If it is, you should see the following IllegalStateException exception in the logs:

java.lang.IllegalStateException: SparkContext has been shutdown
  at org.apache.spark.SparkContext.runJob(SparkContext.scala:1893)
  at org.apache.spark.SparkContext.runJob(SparkContext.scala:1914)
  at org.apache.spark.SparkContext.runJob(SparkContext.scala:1934)
  ... 48 elided

You should see the following INFO message in the logs:

INFO SparkContext: Starting job: [callSite]

With spark.logLineage enabled (which is not by default), you should see the following INFO message with toDebugString (executed on rdd):

INFO SparkContext: RDD's recursive dependencies:
[toDebugString]

runJob requests DAGScheduler to run a job.

Tip
runJob just prepares input parameters for DAGScheduler to run a job.

After DAGScheduler is done and the job has finished, runJob stops ConsoleProgressBar and performs RDD checkpointing of rdd.

Tip
For some actions, e.g. first() and lookup(), there is no need to compute all the partitions of the RDD in a job. And Spark knows it.
// RDD to work with
val lines = sc.parallelize(Seq("hello world", "nice to see you"))

import org.apache.spark.TaskContext
scala> sc.runJob(lines, (t: TaskContext, i: Iterator[String]) => 1) (1)
res0: Array[Int] = Array(1, 1)  (2)
  1. Run a job using runJob on lines RDD with a function that returns 1 for every partition (of lines RDD).

  2. What can you say about the number of partitions of the lines RDD? Is your result res0 different than mine? Why?

Tip
Read TaskContext.

Running a job is essentially executing a func function on all or a subset of partitions in an rdd RDD and returning the result as an array (with elements being the results per partition).

spark runjob.png
Figure 6. Executing action

postApplicationEnd Method

Caution
FIXME

clearActiveContext Method

Caution
FIXME

Stopping SparkContext — stop Method

stop(): Unit

stop stops the SparkContext.

Internally, stop enables stopped internal flag. If already stopped, you should see the following INFO message in the logs:

INFO SparkContext: SparkContext already stopped.

stop then does the following:

  1. Removes _shutdownHookRef from ShutdownHookManager.

  2. Posts a SparkListenerApplicationEnd (to LiveListenerBus Event Bus).

  3. Stops web UI

  4. Requests MetricSystem to report metrics (from all registered sinks).

  5. Stops ContextCleaner.

  6. Requests ExecutorAllocationManager to stop.

  7. If LiveListenerBus was started, requests LiveListenerBus to stop.

  8. Requests EventLoggingListener to stop.

  9. Requests DAGScheduler to stop.

  10. Requests RpcEnv to stop HeartbeatReceiver endpoint.

  11. Requests ConsoleProgressBar to stop.

  12. Clears the reference to TaskScheduler, i.e. _taskScheduler is null.

  13. Requests SparkEnv to stop and clears SparkEnv.

  14. Clears SPARK_YARN_MODE flag.

  15. Clears an active SparkContext.

Ultimately, you should see the following INFO message in the logs:

INFO SparkContext: Successfully stopped SparkContext

Registering SparkListener — addSparkListener Method

addSparkListener(listener: SparkListenerInterface): Unit

You can register a custom SparkListenerInterface using addSparkListener method

Note
You can also register custom listeners using spark.extraListeners setting.

Custom SchedulerBackend, TaskScheduler and DAGScheduler

By default, SparkContext uses (private[spark] class) org.apache.spark.scheduler.DAGScheduler, but you can develop your own custom DAGScheduler implementation, and use (private[spark]) SparkContext.dagScheduler_=(ds: DAGScheduler) method to assign yours.

It is also applicable to SchedulerBackend and TaskScheduler using schedulerBackend_=(sb: SchedulerBackend) and taskScheduler_=(ts: TaskScheduler) methods, respectively.

Caution
FIXME Make it an advanced exercise.

Events

When a Spark context starts, it triggers SparkListenerEnvironmentUpdate and SparkListenerApplicationStart messages.

Refer to the section SparkContext’s initialization.

Setting Default Logging Level — setLogLevel Method

setLogLevel(logLevel: String)

setLogLevel allows you to set the root logging level in a Spark application, e.g. Spark shell.

Tip

You can directly set the logging level using org.apache.log4j.LogManager.getLogger().

LogManager.getLogger("org").setLevel(Level.OFF)

Closure Cleaning — clean Method

clean(f: F, checkSerializable: Boolean = true): F

Every time an action is called, Spark cleans up the closure, i.e. the body of the action, before it is serialized and sent over the wire to executors.

SparkContext comes with clean(f: F, checkSerializable: Boolean = true) method that does this. It in turn calls ClosureCleaner.clean method.

Not only does ClosureCleaner.clean method clean the closure, but also does it transitively, i.e. referenced closures are cleaned transitively.

A closure is considered serializable as long as it does not explicitly reference unserializable objects. It does so by traversing the hierarchy of enclosing closures and null out any references that are not actually used by the starting closure.

Tip

Enable DEBUG logging level for org.apache.spark.util.ClosureCleaner logger to see what happens inside the class.

Add the following line to conf/log4j.properties:

log4j.logger.org.apache.spark.util.ClosureCleaner=DEBUG

Refer to Logging.

With DEBUG logging level you should see the following messages in the logs:

+++ Cleaning closure [func] ([func.getClass.getName]) +++
 + declared fields: [declaredFields.size]
     [field]
 ...
+++ closure [func] ([func.getClass.getName]) is now cleaned +++

Serialization is verified using a new instance of Serializer (as closure Serializer). Refer to Serialization.

Caution
FIXME an example, please.

Hadoop Configuration

While a SparkContext is being created, so is a Hadoop configuration (as an instance of org.apache.hadoop.conf.Configuration that is available as _hadoopConfiguration).

If a SparkConf is provided it is used to build the configuration as described. Otherwise, the default Configuration object is returned.

If AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY are both available, the following settings are set for the Hadoop configuration:

  • fs.s3.awsAccessKeyId, fs.s3n.awsAccessKeyId, fs.s3a.access.key are set to the value of AWS_ACCESS_KEY_ID

  • fs.s3.awsSecretAccessKey, fs.s3n.awsSecretAccessKey, and fs.s3a.secret.key are set to the value of AWS_SECRET_ACCESS_KEY

Every spark.hadoop. setting becomes a setting of the configuration with the prefix spark.hadoop. removed for the key.

The value of spark.buffer.size (default: 65536) is used as the value of io.file.buffer.size.

listenerBus — LiveListenerBus Event Bus

listenerBus is a LiveListenerBus object that acts as a mechanism to announce events to other services on the driver.

Note
It is created and started when SparkContext starts and, since it is a single-JVM event bus, is exclusively used on the driver.
Note
listenerBus is a private[spark] value in SparkContext.

Time when SparkContext was Created — startTime Property

startTime: Long

startTime is the time in milliseconds when SparkContext was created.

scala> sc.startTime
res0: Long = 1464425605653

Spark User — sparkUser Property

sparkUser: String

sparkUser is the user who started the SparkContext instance.

Note
It is computed when SparkContext is created using Utils.getCurrentUserName.

Submitting ShuffleDependency for Execution — submitMapStage Internal Method

submitMapStage[K, V, C](
  dependency: ShuffleDependency[K, V, C]): SimpleFutureAction[MapOutputStatistics]

submitMapStage submits the input ShuffleDependency to DAGScheduler for execution and returns a SimpleFutureAction.

Internally, submitMapStage calculates the call site first and submits it with localProperties.

Note
Interestingly, submitMapStage is used exclusively when Spark SQL’s ShuffleExchange physical operator is executed.
Note
submitMapStage seems related to Adaptive Query Planning / Adaptive Scheduling.

Calculating Call Site — getCallSite Method

Caution
FIXME

cancelJobGroup Method

cancelJobGroup(groupId: String)

cancelAllJobs Method

Caution
FIXME

setJobGroup Method

setJobGroup(
  groupId: String,
  description: String,
  interruptOnCancel: Boolean = false): Unit
Caution
FIXME

cleaner Method

cleaner: Option[ContextCleaner]

cleaner is a private[spark] method to get the optional application-wide ContextCleaner.

Finding Preferred Locations (Placement Preferences) for RDD Partition — getPreferredLocs Method

getPreferredLocs(rdd: RDD[_], partition: Int): Seq[TaskLocation]
Note
Preferred locations of a partition of a RDD are also called placement preferences or locality preferences.
Note
getPreferredLocs is used in CoalescedRDDPartition, DefaultPartitionCoalescer and PartitionerAwareUnionRDD.

Registering RDD in persistentRdds Internal Registry — persistRDD Internal Method

persistRDD(rdd: RDD[_]): Unit

persistRDD registers rdd in persistentRdds internal registry.

Note
persistRDD is used exclusively when RDD is persisted or locally checkpointed.

Settings

spark.driver.allowMultipleContexts

Quoting the scaladoc of org.apache.spark.SparkContext:

Only one SparkContext may be active per JVM. You must stop() the active SparkContext before creating a new one.

You can however control the behaviour using spark.driver.allowMultipleContexts flag.

It is disabled, i.e. false, by default.

If enabled (i.e. true), Spark prints the following WARN message to the logs:

WARN Multiple running SparkContexts detected in the same JVM!

If disabled (default), it will throw an SparkException exception:

Only one SparkContext may be running in this JVM (see SPARK-2243). To ignore this error, set spark.driver.allowMultipleContexts = true. The currently running SparkContext was created at:
[ctx.creationSite.longForm]

When creating an instance of SparkContext, Spark marks the current thread as having it being created (very early in the instantiation process).

Caution
It’s not guaranteed that Spark will work properly with two or more SparkContexts. Consider the feature a work in progress.

Environment Variables

Table 3. Environment Variables
Environment Variable Default Value Description

SPARK_EXECUTOR_MEMORY

1024

Amount of memory to allocate for a Spark executor in MB.

See Executor Memory.

SPARK_USER

The user who is running SparkContext. Available later as sparkUser.

results matching ""

    No results matching ""