SparkContext

addFile adds the path file to be downloaded…​FIXME

unpersistRDD requests BlockManagerMaster to remove the blocks for the RDD (given rddId).

unpersistRDD removes rddId from persistentRdds registry.

In the end, unpersistRDD posts a SparkListenerUnpersistRDD (with rddId) to LiveListenerBus Event Bus.

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

Internally, getPersistentRDDs returns persistentRdds internal registry.

cancelJob requests DAGScheduler to cancel a Spark job.

cancelStage simply requests DAGScheduler to cancel a Spark stage (with an optional reason).

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

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

getConf returns the current SparkConf.

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

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

applicationAttemptId gives the unique identifier of the execution attempt of a Spark application.

getExecutorStorageStatus requests BlockManagerMaster for storage status (of all BlockManagers).

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

getSchedulingMode returns the current Scheduling Mode.

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

Internally, it requests the TaskScheduler for the root pool and looks up the Schedulable by the pool name.

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

getAllPools collects the Pools in TaskScheduler.rootPool.

defaultParallelism requests TaskScheduler for the default level of parallelism.

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

version returns the Spark version this SparkContext uses.

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.

It is used in:

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 allows you to create many different RDDs from input sources like:

Read Creating RDDs in RDD – Resilient Distributed Dataset.

unpersist 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.

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

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

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

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].

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).

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.

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.

Internally, broadcast requests the current BroadcastManager to create a new broadcast variable.

You should see the following INFO message in the logs:

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

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

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.

SparkContext keeps track of:

RDD actions run jobs using one of runJob methods.

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

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

runJob then calculates the call site and cleans a func closure.

You should see the following INFO message in the logs:

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

runJob requests DAGScheduler to run a job.

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

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).

stop stops the SparkContext.

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

stop then does the following:

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

You can register a custom SparkListenerInterface using addSparkListener method

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.

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

Refer to the section SparkContext’s initialization.

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

Internally, setLogLevel calls org.apache.log4j.Level.toLevel(logLevel) that it then uses to set using org.apache.log4j.LogManager.getRootLogger().setLevel(level).

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.

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

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

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:

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 is a LiveListenerBus object that acts as a mechanism to announce events to other services on the driver.

startTime is the time in milliseconds when SparkContext was created.

sparkUser is the user who started the SparkContext instance.

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.

cancelJobGroup requests DAGScheduler to cancel a group of active Spark jobs.

setJobGroup sets local properties:

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

getPreferredLocs simply requests DAGScheduler for the preferred locations for partition.

persistRDD registers rdd in persistentRdds internal registry.

getRDDStorageInfo takes all the RDDs (from persistentRdds registry) that match filter and creates a collection of RDDInfo instances.

getRDDStorageInfo then updates the RDDInfos with the current status of all BlockManagers (in a Spark application).

In the end, getRDDStorageInfo gives only the RDD that are cached (i.e. the sum of memory and disk sizes as well as the number of partitions cached are greater than 0).

statusStore gives the current AppStatusStore.

uiWebUrl requests the SparkUI for webUrl.