class SparkContext extends Logging
Main entry point for Spark functionality. A SparkContext represents the connection to a Spark cluster, and can be used to create RDDs, accumulators and broadcast variables on that cluster.
- Source
- SparkContext.scala
- Note
Only one
SparkContext
should be active per JVM. You muststop()
the activeSparkContext
before creating a new one.
- Alphabetic
- By Inheritance
- SparkContext
- Logging
- AnyRef
- Any
- Hide All
- Show All
- Public
- All
Instance Constructors
-
new
SparkContext(master: String, appName: String, sparkHome: String = null, jars: Seq[String] = Nil, environment: Map[String, String] = Map())
Alternative constructor that allows setting common Spark properties directly
Alternative constructor that allows setting common Spark properties directly
- master
Cluster URL to connect to (e.g. mesos://host:port, spark://host:port, local[4]).
- appName
A name for your application, to display on the cluster web UI.
- sparkHome
Location where Spark is installed on cluster nodes.
- jars
Collection of JARs to send to the cluster. These can be paths on the local file system or HDFS, HTTP, HTTPS, or FTP URLs.
- environment
Environment variables to set on worker nodes.
-
new
SparkContext(master: String, appName: String, conf: SparkConf)
Alternative constructor that allows setting common Spark properties directly
Alternative constructor that allows setting common Spark properties directly
- master
Cluster URL to connect to (e.g. mesos://host:port, spark://host:port, local[4]).
- appName
A name for your application, to display on the cluster web UI
- conf
a org.apache.spark.SparkConf object specifying other Spark parameters
-
new
SparkContext()
Create a SparkContext that loads settings from system properties (for instance, when launching with ./bin/spark-submit).
-
new
SparkContext(config: SparkConf)
- config
a Spark Config object describing the application configuration. Any settings in this config overrides the default configs as well as system properties.
Value Members
-
final
def
!=(arg0: Any): Boolean
- Definition Classes
- AnyRef → Any
-
final
def
##(): Int
- Definition Classes
- AnyRef → Any
-
final
def
==(arg0: Any): Boolean
- Definition Classes
- AnyRef → Any
-
def
addArchive(path: String): Unit
:: Experimental :: Add an archive to be downloaded and unpacked with this Spark job on every node.
:: Experimental :: Add an archive to be downloaded and unpacked with this Spark job on every node.
If an archive is added during execution, it will not be available until the next TaskSet starts.
- path
can be either a local file, a file in HDFS (or other Hadoop-supported filesystems), or an HTTP, HTTPS or FTP URI. To access the file in Spark jobs, use
SparkFiles.get(paths-to-files)
to find its download/unpacked location. The given path should be one of .zip, .tar, .tar.gz, .tgz and .jar.
- Annotations
- @Experimental()
- Since
3.1.0
- Note
A path can be added only once. Subsequent additions of the same path are ignored.
-
def
addFile(path: String, recursive: Boolean): Unit
Add a file to be downloaded with this Spark job on every node.
Add a file to be downloaded with this Spark job on every node.
If a file is added during execution, it will not be available until the next TaskSet starts.
- path
can be either a local file, a file in HDFS (or other Hadoop-supported filesystems), or an HTTP, HTTPS or FTP URI. To access the file in Spark jobs, use
SparkFiles.get(fileName)
to find its download location.- recursive
if true, a directory can be given in
path
. Currently directories are only supported for Hadoop-supported filesystems.
- Note
A path can be added only once. Subsequent additions of the same path are ignored.
-
def
addFile(path: String): Unit
Add a file to be downloaded with this Spark job on every node.
Add a file to be downloaded with this Spark job on every node.
If a file is added during execution, it will not be available until the next TaskSet starts.
- path
can be either a local file, a file in HDFS (or other Hadoop-supported filesystems), or an HTTP, HTTPS or FTP URI. To access the file in Spark jobs, use
SparkFiles.get(fileName)
to find its download location.
- Note
A path can be added only once. Subsequent additions of the same path are ignored.
-
def
addJar(path: String): Unit
Adds a JAR dependency for all tasks to be executed on this
SparkContext
in the future.Adds a JAR dependency for all tasks to be executed on this
SparkContext
in the future.If a jar is added during execution, it will not be available until the next TaskSet starts.
- path
can be either 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.
- Note
A path can be added only once. Subsequent additions of the same path are ignored.
-
def
addJobTag(tag: String): Unit
Add a tag to be assigned to all the jobs started by this thread.
Add a tag to be assigned to all the jobs started by this thread.
- tag
The tag to be added. Cannot contain ',' (comma) character.
- Since
3.5.0
-
def
addSparkListener(listener: SparkListenerInterface): Unit
:: DeveloperApi :: Register a listener to receive up-calls from events that happen during execution.
:: DeveloperApi :: Register a listener to receive up-calls from events that happen during execution.
- Annotations
- @DeveloperApi()
- def appName: String
- def applicationAttemptId: Option[String]
-
def
applicationId: String
A unique identifier for the Spark application.
A unique identifier for the Spark application. Its format depends on the scheduler implementation. (i.e. in case of local spark app something like 'local-1433865536131' in case of YARN something like 'application_1433865536131_34483' in case of MESOS something like 'driver-20170926223339-0001' )
- def archives: Seq[String]
-
final
def
asInstanceOf[T0]: T0
- Definition Classes
- Any
-
def
binaryFiles(path: String, minPartitions: Int = defaultMinPartitions): RDD[(String, PortableDataStream)]
Get an RDD for a Hadoop-readable dataset as PortableDataStream for each file (useful for binary data)
Get an RDD for a Hadoop-readable dataset as PortableDataStream for each file (useful for binary data)
For example, if you have the following files:
hdfs://a-hdfs-path/part-00000 hdfs://a-hdfs-path/part-00001 ... hdfs://a-hdfs-path/part-nnnnn
Do
val rdd = sparkContext.binaryFiles("hdfs://a-hdfs-path")
,then
rdd
contains(a-hdfs-path/part-00000, its content) (a-hdfs-path/part-00001, its content) ... (a-hdfs-path/part-nnnnn, its content)
- path
Directory to the input data files, the path can be comma separated paths as the list of inputs.
- minPartitions
A suggestion value of the minimal splitting number for input data.
- returns
RDD representing tuples of file path and corresponding file content
- Note
Small files are preferred; very large files may cause bad performance.
,On some filesystems,
,.../path/*
can be a more efficient way to read all files in a directory rather than.../path/
or.../path
Partitioning is determined by data locality. This may result in too few partitions by default.
-
def
binaryRecords(path: String, recordLength: Int, conf: Configuration = hadoopConfiguration): RDD[Array[Byte]]
Load data from a flat binary file, assuming the length of each record is constant.
Load data from a flat binary file, assuming the length of each record is constant.
- path
Directory to the input data files, the path can be comma separated paths as the list of inputs.
- recordLength
The length at which to split the records
- conf
Configuration for setting up the dataset.
- returns
An RDD of data with values, represented as byte arrays
- Note
We ensure that the byte array for each record in the resulting RDD has the provided record length.
-
def
broadcast[T](value: T)(implicit arg0: ClassTag[T]): Broadcast[T]
Broadcast a read-only variable to the cluster, returning a org.apache.spark.broadcast.Broadcast object for reading it in distributed functions.
Broadcast a read-only variable to the cluster, returning a org.apache.spark.broadcast.Broadcast object for reading it in distributed functions. The variable will be sent to each executor only once.
- value
value to broadcast to the Spark nodes
- returns
Broadcast
object, a read-only variable cached on each machine
-
def
cancelAllJobs(): Unit
Cancel all jobs that have been scheduled or are running.
-
def
cancelJob(jobId: Int): Unit
Cancel a given job if it's scheduled or running.
Cancel a given job if it's scheduled or running.
- jobId
the job ID to cancel
- Note
Throws
InterruptedException
if the cancel message cannot be sent
-
def
cancelJob(jobId: Int, reason: String): Unit
Cancel a given job if it's scheduled or running.
Cancel a given job if it's scheduled or running.
- jobId
the job ID to cancel
- reason
optional reason for cancellation
- Note
Throws
InterruptedException
if the cancel message cannot be sent
-
def
cancelJobGroup(groupId: String): Unit
Cancel active jobs for the specified group.
Cancel active jobs for the specified group. See
org.apache.spark.SparkContext.setJobGroup
for more information. -
def
cancelJobsWithTag(tag: String): Unit
Cancel active jobs that have the specified tag.
Cancel active jobs that have the specified tag. See
org.apache.spark.SparkContext.addJobTag
.- tag
The tag to be cancelled. Cannot contain ',' (comma) character.
- Since
3.5.0
-
def
cancelStage(stageId: Int): Unit
Cancel a given stage and all jobs associated with it.
Cancel a given stage and all jobs associated with it.
- stageId
the stage ID to cancel
- Note
Throws
InterruptedException
if the cancel message cannot be sent
-
def
cancelStage(stageId: Int, reason: String): Unit
Cancel a given stage and all jobs associated with it.
Cancel a given stage and all jobs associated with it.
- stageId
the stage ID to cancel
- reason
reason for cancellation
- Note
Throws
InterruptedException
if the cancel message cannot be sent
-
def
checkpointFile[T](path: String)(implicit arg0: ClassTag[T]): RDD[T]
- Attributes
- protected[spark]
-
def
clearCallSite(): Unit
Clear the thread-local property for overriding the call sites of actions and RDDs.
-
def
clearJobGroup(): Unit
Clear the current thread's job group ID and its description.
-
def
clearJobTags(): Unit
Clear the current thread's job tags.
Clear the current thread's job tags.
- Since
3.5.0
-
def
clone(): AnyRef
- Attributes
- protected[lang]
- Definition Classes
- AnyRef
- Annotations
- @throws( ... ) @native()
-
def
collectionAccumulator[T](name: String): CollectionAccumulator[T]
Create and register a
CollectionAccumulator
, which starts with empty list and accumulates inputs by adding them into the list. -
def
collectionAccumulator[T]: CollectionAccumulator[T]
Create and register a
CollectionAccumulator
, which starts with empty list and accumulates inputs by adding them into the list. -
def
defaultMinPartitions: Int
Default min number of partitions for Hadoop RDDs when not given by user Notice that we use math.min so the "defaultMinPartitions" cannot be higher than 2.
Default min number of partitions for Hadoop RDDs when not given by user Notice that we use math.min so the "defaultMinPartitions" cannot be higher than 2. The reasons for this are discussed in https://github.com/mesos/spark/pull/718
-
def
defaultParallelism: Int
Default level of parallelism to use when not given by user (e.g.
Default level of parallelism to use when not given by user (e.g. parallelize and makeRDD).
- def deployMode: String
-
def
doubleAccumulator(name: String): DoubleAccumulator
Create and register a double accumulator, which starts with 0 and accumulates inputs by
add
. -
def
doubleAccumulator: DoubleAccumulator
Create and register a double accumulator, which starts with 0 and accumulates inputs by
add
. -
def
emptyRDD[T](implicit arg0: ClassTag[T]): RDD[T]
Get an RDD that has no partitions or elements.
-
final
def
eq(arg0: AnyRef): Boolean
- Definition Classes
- AnyRef
-
def
equals(arg0: Any): Boolean
- Definition Classes
- AnyRef → Any
- def files: Seq[String]
-
def
finalize(): Unit
- Attributes
- protected[lang]
- Definition Classes
- AnyRef
- Annotations
- @throws( classOf[java.lang.Throwable] )
-
def
getAllPools: Seq[Schedulable]
:: DeveloperApi :: Return pools for fair scheduler
:: DeveloperApi :: Return pools for fair scheduler
- Annotations
- @DeveloperApi()
- def getCheckpointDir: Option[String]
-
final
def
getClass(): Class[_]
- Definition Classes
- AnyRef → Any
- Annotations
- @native()
-
def
getConf: SparkConf
Return a copy of this SparkContext's configuration.
Return a copy of this SparkContext's configuration. The configuration cannot be changed at runtime.
-
def
getExecutorMemoryStatus: Map[String, (Long, Long)]
Return a map from the block manager to the max memory available for caching and the remaining memory available for caching.
-
def
getJobTags(): Set[String]
Get the tags that are currently set to be assigned to all the jobs started by this thread.
Get the tags that are currently set to be assigned to all the jobs started by this thread.
- Since
3.5.0
-
def
getLocalProperty(key: String): String
Get a local property set in this thread, or null if it is missing.
Get a local property set in this thread, or null if it is missing. See
org.apache.spark.SparkContext.setLocalProperty
. -
def
getPersistentRDDs: Map[Int, RDD[_]]
Returns an immutable map of RDDs that have marked themselves as persistent via cache() call.
Returns an immutable map of RDDs that have marked themselves as persistent via cache() call.
- Note
This does not necessarily mean the caching or computation was successful.
-
def
getPoolForName(pool: String): Option[Schedulable]
:: DeveloperApi :: Return the pool associated with the given name, if one exists
:: DeveloperApi :: Return the pool associated with the given name, if one exists
- Annotations
- @DeveloperApi()
-
def
getRDDStorageInfo: Array[RDDInfo]
:: DeveloperApi :: Return information about what RDDs are cached, if they are in mem or on disk, how much space they take, etc.
:: DeveloperApi :: Return information about what RDDs are cached, if they are in mem or on disk, how much space they take, etc.
- Annotations
- @DeveloperApi()
-
def
getSchedulingMode: SchedulingMode
Return current scheduling mode
-
def
hadoopConfiguration: Configuration
A default Hadoop Configuration for the Hadoop code (e.g.
A default Hadoop Configuration for the Hadoop code (e.g. file systems) that we reuse.
- Note
As it will be reused in all Hadoop RDDs, it's better not to modify it unless you plan to set some global configurations for all Hadoop RDDs.
-
def
hadoopFile[K, V, F <: InputFormat[K, V]](path: String)(implicit km: ClassTag[K], vm: ClassTag[V], fm: ClassTag[F]): RDD[(K, V)]
Smarter version of hadoopFile() that uses class tags to figure out the classes of keys, values and the InputFormat so that users don't need to pass them directly.
Smarter version of hadoopFile() that uses class tags to figure out the classes of keys, values and the InputFormat so that users don't need to pass them directly. Instead, callers can just write, for example,
val file = sparkContext.hadoopFile[LongWritable, Text, TextInputFormat](path)
- path
directory to the input data files, the path can be comma separated paths as a list of inputs
- returns
RDD of tuples of key and corresponding value
- Note
Because Hadoop's RecordReader class re-uses the same Writable object for each record, directly caching the returned RDD or directly passing it to an aggregation or shuffle operation will create many references to the same object. If you plan to directly cache, sort, or aggregate Hadoop writable objects, you should first copy them using a
map
function.
-
def
hadoopFile[K, V, F <: InputFormat[K, V]](path: String, minPartitions: Int)(implicit km: ClassTag[K], vm: ClassTag[V], fm: ClassTag[F]): RDD[(K, V)]
Smarter version of hadoopFile() that uses class tags to figure out the classes of keys, values and the InputFormat so that users don't need to pass them directly.
Smarter version of hadoopFile() that uses class tags to figure out the classes of keys, values and the InputFormat so that users don't need to pass them directly. Instead, callers can just write, for example,
val file = sparkContext.hadoopFile[LongWritable, Text, TextInputFormat](path, minPartitions)
- path
directory to the input data files, the path can be comma separated paths as a list of inputs
- minPartitions
suggested minimum number of partitions for the resulting RDD
- returns
RDD of tuples of key and corresponding value
- Note
Because Hadoop's RecordReader class re-uses the same Writable object for each record, directly caching the returned RDD or directly passing it to an aggregation or shuffle operation will create many references to the same object. If you plan to directly cache, sort, or aggregate Hadoop writable objects, you should first copy them using a
map
function.
-
def
hadoopFile[K, V](path: String, inputFormatClass: Class[_ <: InputFormat[K, V]], keyClass: Class[K], valueClass: Class[V], minPartitions: Int = defaultMinPartitions): RDD[(K, V)]
Get an RDD for a Hadoop file with an arbitrary InputFormat
Get an RDD for a Hadoop file with an arbitrary InputFormat
- path
directory to the input data files, the path can be comma separated paths as a list of inputs
- inputFormatClass
storage format of the data to be read
- keyClass
Class
of the key associated with theinputFormatClass
parameter- valueClass
Class
of the value associated with theinputFormatClass
parameter- minPartitions
suggested minimum number of partitions for the resulting RDD
- returns
RDD of tuples of key and corresponding value
- Note
Because Hadoop's RecordReader class re-uses the same Writable object for each record, directly caching the returned RDD or directly passing it to an aggregation or shuffle operation will create many references to the same object. If you plan to directly cache, sort, or aggregate Hadoop writable objects, you should first copy them using a
map
function.
-
def
hadoopRDD[K, V](conf: JobConf, inputFormatClass: Class[_ <: InputFormat[K, V]], keyClass: Class[K], valueClass: Class[V], minPartitions: Int = defaultMinPartitions): RDD[(K, V)]
Get an RDD for a Hadoop-readable dataset from a Hadoop JobConf given its InputFormat and other necessary info (e.g.
Get an RDD for a Hadoop-readable dataset from a Hadoop JobConf given its InputFormat and other necessary info (e.g. file name for a filesystem-based dataset, table name for HyperTable), using the older MapReduce API (
org.apache.hadoop.mapred
).- conf
JobConf for setting up the dataset. Note: This will be put into a Broadcast. Therefore if you plan to reuse this conf to create multiple RDDs, you need to make sure you won't modify the conf. A safe approach is always creating a new conf for a new RDD.
- inputFormatClass
storage format of the data to be read
- keyClass
Class
of the key associated with theinputFormatClass
parameter- valueClass
Class
of the value associated with theinputFormatClass
parameter- minPartitions
Minimum number of Hadoop Splits to generate.
- returns
RDD of tuples of key and corresponding value
- Note
Because Hadoop's RecordReader class re-uses the same Writable object for each record, directly caching the returned RDD or directly passing it to an aggregation or shuffle operation will create many references to the same object. If you plan to directly cache, sort, or aggregate Hadoop writable objects, you should first copy them using a
map
function.
-
def
hashCode(): Int
- Definition Classes
- AnyRef → Any
- Annotations
- @native()
-
def
initializeLogIfNecessary(isInterpreter: Boolean, silent: Boolean): Boolean
- Attributes
- protected
- Definition Classes
- Logging
-
def
initializeLogIfNecessary(isInterpreter: Boolean): Unit
- Attributes
- protected
- Definition Classes
- Logging
-
final
def
isInstanceOf[T0]: Boolean
- Definition Classes
- Any
- def isLocal: Boolean
-
def
isStopped: Boolean
- returns
true if context is stopped or in the midst of stopping.
-
def
isTraceEnabled(): Boolean
- Attributes
- protected
- Definition Classes
- Logging
- def jars: Seq[String]
-
def
killExecutor(executorId: String): Boolean
:: DeveloperApi :: Request that the cluster manager kill the specified executor.
:: DeveloperApi :: Request that the cluster manager kill the specified executor.
- returns
whether the request is received.
- Annotations
- @DeveloperApi()
- Note
This is an indication to the cluster manager that the application wishes to adjust its resource usage downwards. If the application wishes to replace the executor it kills through this method with a new one, it should follow up explicitly with a call to {{SparkContext#requestExecutors}}.
-
def
killExecutors(executorIds: Seq[String]): Boolean
:: DeveloperApi :: Request that the cluster manager kill the specified executors.
:: DeveloperApi :: Request that the cluster manager kill the specified executors.
This is not supported when dynamic allocation is turned on.
- returns
whether the request is received.
- Annotations
- @DeveloperApi()
- Note
This is an indication to the cluster manager that the application wishes to adjust its resource usage downwards. If the application wishes to replace the executors it kills through this method with new ones, it should follow up explicitly with a call to {{SparkContext#requestExecutors}}.
-
def
killTaskAttempt(taskId: Long, interruptThread: Boolean = true, reason: String = ...): Boolean
Kill and reschedule the given task attempt.
Kill and reschedule the given task attempt. Task ids can be obtained from the Spark UI or through SparkListener.onTaskStart.
- taskId
the task ID to kill. This id uniquely identifies the task attempt.
- interruptThread
whether to interrupt the thread running the task.
- reason
the reason for killing the task, which should be a short string. If a task is killed multiple times with different reasons, only one reason will be reported.
- returns
Whether the task was successfully killed.
-
def
listArchives(): Seq[String]
:: Experimental :: Returns a list of archive paths that are added to resources.
:: Experimental :: Returns a list of archive paths that are added to resources.
- Annotations
- @Experimental()
- Since
3.1.0
-
def
listFiles(): Seq[String]
Returns a list of file paths that are added to resources.
-
def
listJars(): Seq[String]
Returns a list of jar files that are added to resources.
-
val
localProperties: InheritableThreadLocal[Properties]
- Attributes
- protected[spark]
-
def
log: Logger
- Attributes
- protected
- Definition Classes
- Logging
-
def
logDebug(msg: ⇒ String, throwable: Throwable): Unit
- Attributes
- protected
- Definition Classes
- Logging
-
def
logDebug(msg: ⇒ String): Unit
- Attributes
- protected
- Definition Classes
- Logging
-
def
logError(msg: ⇒ String, throwable: Throwable): Unit
- Attributes
- protected
- Definition Classes
- Logging
-
def
logError(msg: ⇒ String): Unit
- Attributes
- protected
- Definition Classes
- Logging
-
def
logInfo(msg: ⇒ String, throwable: Throwable): Unit
- Attributes
- protected
- Definition Classes
- Logging
-
def
logInfo(msg: ⇒ String): Unit
- Attributes
- protected
- Definition Classes
- Logging
-
def
logName: String
- Attributes
- protected
- Definition Classes
- Logging
-
def
logTrace(msg: ⇒ String, throwable: Throwable): Unit
- Attributes
- protected
- Definition Classes
- Logging
-
def
logTrace(msg: ⇒ String): Unit
- Attributes
- protected
- Definition Classes
- Logging
-
def
logWarning(msg: ⇒ String, throwable: Throwable): Unit
- Attributes
- protected
- Definition Classes
- Logging
-
def
logWarning(msg: ⇒ String): Unit
- Attributes
- protected
- Definition Classes
- Logging
-
def
longAccumulator(name: String): LongAccumulator
Create and register a long accumulator, which starts with 0 and accumulates inputs by
add
. -
def
longAccumulator: LongAccumulator
Create and register a long accumulator, which starts with 0 and accumulates inputs by
add
. -
def
makeRDD[T](seq: Seq[(T, Seq[String])])(implicit arg0: ClassTag[T]): RDD[T]
Distribute a local Scala collection to form an RDD, with one or more location preferences (hostnames of Spark nodes) for each object.
Distribute a local Scala collection to form an RDD, with one or more location preferences (hostnames of Spark nodes) for each object. Create a new partition for each collection item.
- seq
list of tuples of data and location preferences (hostnames of Spark nodes)
- returns
RDD representing data partitioned according to location preferences
-
def
makeRDD[T](seq: Seq[T], numSlices: Int = defaultParallelism)(implicit arg0: ClassTag[T]): RDD[T]
Distribute a local Scala collection to form an RDD.
Distribute a local Scala collection to form an RDD.
This method is identical to
parallelize
.- seq
Scala collection to distribute
- numSlices
number of partitions to divide the collection into
- returns
RDD representing distributed collection
- def master: String
-
final
def
ne(arg0: AnyRef): Boolean
- Definition Classes
- AnyRef
-
def
newAPIHadoopFile[K, V, F <: InputFormat[K, V]](path: String, fClass: Class[F], kClass: Class[K], vClass: Class[V], conf: Configuration = hadoopConfiguration): RDD[(K, V)]
Get an RDD for a given Hadoop file with an arbitrary new API InputFormat and extra configuration options to pass to the input format.
Get an RDD for a given Hadoop file with an arbitrary new API InputFormat and extra configuration options to pass to the input format.
- path
directory to the input data files, the path can be comma separated paths as a list of inputs
- fClass
storage format of the data to be read
- kClass
Class
of the key associated with thefClass
parameter- vClass
Class
of the value associated with thefClass
parameter- conf
Hadoop configuration
- returns
RDD of tuples of key and corresponding value
- Note
Because Hadoop's RecordReader class re-uses the same Writable object for each record, directly caching the returned RDD or directly passing it to an aggregation or shuffle operation will create many references to the same object. If you plan to directly cache, sort, or aggregate Hadoop writable objects, you should first copy them using a
map
function.
-
def
newAPIHadoopFile[K, V, F <: InputFormat[K, V]](path: String)(implicit km: ClassTag[K], vm: ClassTag[V], fm: ClassTag[F]): RDD[(K, V)]
Smarter version of
newApiHadoopFile
that uses class tags to figure out the classes of keys, values and theorg.apache.hadoop.mapreduce.InputFormat
(new MapReduce API) so that user don't need to pass them directly.Smarter version of
newApiHadoopFile
that uses class tags to figure out the classes of keys, values and theorg.apache.hadoop.mapreduce.InputFormat
(new MapReduce API) so that user don't need to pass them directly. Instead, callers can just write, for example:val file = sparkContext.hadoopFile[LongWritable, Text, TextInputFormat](path)
- path
directory to the input data files, the path can be comma separated paths as a list of inputs
- returns
RDD of tuples of key and corresponding value
- Note
Because Hadoop's RecordReader class re-uses the same Writable object for each record, directly caching the returned RDD or directly passing it to an aggregation or shuffle operation will create many references to the same object. If you plan to directly cache, sort, or aggregate Hadoop writable objects, you should first copy them using a
map
function.
-
def
newAPIHadoopRDD[K, V, F <: InputFormat[K, V]](conf: Configuration = hadoopConfiguration, fClass: Class[F], kClass: Class[K], vClass: Class[V]): RDD[(K, V)]
Get an RDD for a given Hadoop file with an arbitrary new API InputFormat and extra configuration options to pass to the input format.
Get an RDD for a given Hadoop file with an arbitrary new API InputFormat and extra configuration options to pass to the input format.
- conf
Configuration for setting up the dataset. Note: This will be put into a Broadcast. Therefore if you plan to reuse this conf to create multiple RDDs, you need to make sure you won't modify the conf. A safe approach is always creating a new conf for a new RDD.
- fClass
storage format of the data to be read
- kClass
Class
of the key associated with thefClass
parameter- vClass
Class
of the value associated with thefClass
parameter
- Note
Because Hadoop's RecordReader class re-uses the same Writable object for each record, directly caching the returned RDD or directly passing it to an aggregation or shuffle operation will create many references to the same object. If you plan to directly cache, sort, or aggregate Hadoop writable objects, you should first copy them using a
map
function.
-
final
def
notify(): Unit
- Definition Classes
- AnyRef
- Annotations
- @native()
-
final
def
notifyAll(): Unit
- Definition Classes
- AnyRef
- Annotations
- @native()
-
def
objectFile[T](path: String, minPartitions: Int = defaultMinPartitions)(implicit arg0: ClassTag[T]): RDD[T]
Load an RDD saved as a SequenceFile containing serialized objects, with NullWritable keys and BytesWritable values that contain a serialized partition.
Load an RDD saved as a SequenceFile containing serialized objects, with NullWritable keys and BytesWritable values that contain a serialized partition. This is still an experimental storage format and may not be supported exactly as is in future Spark releases. It will also be pretty slow if you use the default serializer (Java serialization), though the nice thing about it is that there's very little effort required to save arbitrary objects.
- path
directory to the input data files, the path can be comma separated paths as a list of inputs
- minPartitions
suggested minimum number of partitions for the resulting RDD
- returns
RDD representing deserialized data from the file(s)
-
def
parallelize[T](seq: Seq[T], numSlices: Int = defaultParallelism)(implicit arg0: ClassTag[T]): RDD[T]
Distribute a local Scala collection to form an RDD.
Distribute a local Scala collection to form an RDD.
- seq
Scala collection to distribute
- numSlices
number of partitions to divide the collection into
- returns
RDD representing distributed collection
- Note
Parallelize acts lazily. If
,seq
is a mutable collection and is altered after the call to parallelize and before the first action on the RDD, the resultant RDD will reflect the modified collection. Pass a copy of the argument to avoid this.avoid using
parallelize(Seq())
to create an emptyRDD
. ConsideremptyRDD
for an RDD with no partitions, orparallelize(Seq[T]())
for an RDD ofT
with empty partitions.
-
def
range(start: Long, end: Long, step: Long = 1, numSlices: Int = defaultParallelism): RDD[Long]
Creates a new RDD[Long] containing elements from
start
toend
(exclusive), increased bystep
every element.Creates a new RDD[Long] containing elements from
start
toend
(exclusive), increased bystep
every element.- start
the start value.
- end
the end value.
- step
the incremental step
- numSlices
number of partitions to divide the collection into
- returns
RDD representing distributed range
- Note
if we need to cache this RDD, we should make sure each partition does not exceed limit.
-
def
register(acc: AccumulatorV2[_, _], name: String): Unit
Register the given accumulator with given name.
Register the given accumulator with given name.
- Note
Accumulators must be registered before use, or it will throw exception.
-
def
register(acc: AccumulatorV2[_, _]): Unit
Register the given accumulator.
Register the given accumulator.
- Note
Accumulators must be registered before use, or it will throw exception.
-
def
removeJobTag(tag: String): Unit
Remove a tag previously added to be assigned to all the jobs started by this thread.
Remove a tag previously added to be assigned to all the jobs started by this thread. Noop if such a tag was not added earlier.
- tag
The tag to be removed. Cannot contain ',' (comma) character.
- Since
3.5.0
-
def
removeSparkListener(listener: SparkListenerInterface): Unit
:: DeveloperApi :: Deregister the listener from Spark's listener bus.
:: DeveloperApi :: Deregister the listener from Spark's listener bus.
- Annotations
- @DeveloperApi()
-
def
requestExecutors(numAdditionalExecutors: Int): Boolean
:: DeveloperApi :: Request an additional number of executors from the cluster manager.
:: DeveloperApi :: Request an additional number of executors from the cluster manager.
- returns
whether the request is received.
- Annotations
- @DeveloperApi()
-
def
requestTotalExecutors(numExecutors: Int, localityAwareTasks: Int, hostToLocalTaskCount: Map[String, Int]): Boolean
Update the cluster manager on our scheduling needs.
Update the cluster manager on our scheduling needs. Three bits of information are included to help it make decisions. This applies to the default ResourceProfile.
- numExecutors
The total number of executors we'd like to have. The cluster manager shouldn't kill any running executor to reach this number, but, if all existing executors were to die, this is the number of executors we'd want to be allocated.
- localityAwareTasks
The number of tasks in all active stages that have a locality preferences. This includes running, pending, and completed tasks.
- hostToLocalTaskCount
A map of hosts to the number of tasks from all active stages that would like to like to run on that host. This includes running, pending, and completed tasks.
- returns
whether the request is acknowledged by the cluster manager.
- Annotations
- @DeveloperApi()
- def resources: Map[String, ResourceInformation]
-
def
runApproximateJob[T, U, R](rdd: RDD[T], func: (TaskContext, Iterator[T]) ⇒ U, evaluator: ApproximateEvaluator[U, R], timeout: Long): PartialResult[R]
:: DeveloperApi :: Run a job that can return approximate results.
:: DeveloperApi :: Run a job that can return approximate results.
- rdd
target RDD to run tasks on
- func
a function to run on each partition of the RDD
- evaluator
ApproximateEvaluator
to receive the partial results- timeout
maximum time to wait for the job, in milliseconds
- returns
partial result (how partial depends on whether the job was finished before or after timeout)
- Annotations
- @DeveloperApi()
-
def
runJob[T, U](rdd: RDD[T], processPartition: (Iterator[T]) ⇒ U, resultHandler: (Int, U) ⇒ Unit)(implicit arg0: ClassTag[U]): Unit
Run a job on all partitions in an RDD and pass the results to a handler function.
Run a job on all partitions in an RDD and pass the results to a handler function.
- rdd
target RDD to run tasks on
- processPartition
a function to run on each partition of the RDD
- resultHandler
callback to pass each result to
-
def
runJob[T, U](rdd: RDD[T], processPartition: (TaskContext, Iterator[T]) ⇒ U, resultHandler: (Int, U) ⇒ Unit)(implicit arg0: ClassTag[U]): Unit
Run a job on all partitions in an RDD and pass the results to a handler function.
Run a job on all partitions in an RDD and pass the results to a handler function. The function that is run against each partition additionally takes
TaskContext
argument.- rdd
target RDD to run tasks on
- processPartition
a function to run on each partition of the RDD
- resultHandler
callback to pass each result to
-
def
runJob[T, U](rdd: RDD[T], func: (Iterator[T]) ⇒ U)(implicit arg0: ClassTag[U]): Array[U]
Run a job on all partitions in an RDD and return the results in an array.
Run a job on all partitions in an RDD and return the results in an array.
- rdd
target RDD to run tasks on
- func
a function to run on each partition of the RDD
- returns
in-memory collection with a result of the job (each collection element will contain a result from one partition)
-
def
runJob[T, U](rdd: RDD[T], func: (TaskContext, Iterator[T]) ⇒ U)(implicit arg0: ClassTag[U]): Array[U]
Run a job on all partitions in an RDD and return the results in an array.
Run a job on all partitions in an RDD and return the results in an array. The function that is run against each partition additionally takes
TaskContext
argument.- rdd
target RDD to run tasks on
- func
a function to run on each partition of the RDD
- returns
in-memory collection with a result of the job (each collection element will contain a result from one partition)
-
def
runJob[T, U](rdd: RDD[T], func: (Iterator[T]) ⇒ U, partitions: Seq[Int])(implicit arg0: ClassTag[U]): Array[U]
Run a function on a given set of partitions in an RDD and return the results as an array.
Run a function on a given set of partitions in an RDD and return the results as an array.
- rdd
target RDD to run tasks on
- func
a function to run on each partition of the RDD
- partitions
set of partitions to run on; some jobs may not want to compute on all partitions of the target RDD, e.g. for operations like
first()
- returns
in-memory collection with a result of the job (each collection element will contain a result from one partition)
-
def
runJob[T, U](rdd: RDD[T], func: (TaskContext, Iterator[T]) ⇒ U, partitions: Seq[Int])(implicit arg0: ClassTag[U]): Array[U]
Run a function on a given set of partitions in an RDD and return the results as an array.
Run a function on a given set of partitions in an RDD and return the results as an array. The function that is run against each partition additionally takes
TaskContext
argument.- rdd
target RDD to run tasks on
- func
a function to run on each partition of the RDD
- partitions
set of partitions to run on; some jobs may not want to compute on all partitions of the target RDD, e.g. for operations like
first()
- returns
in-memory collection with a result of the job (each collection element will contain a result from one partition)
-
def
runJob[T, U](rdd: RDD[T], func: (TaskContext, Iterator[T]) ⇒ U, partitions: Seq[Int], resultHandler: (Int, U) ⇒ Unit)(implicit arg0: ClassTag[U]): Unit
Run a function on a given set of partitions in an RDD and pass the results to the given handler function.
Run a function on a given set of partitions in an RDD and pass the results to the given handler function. This is the main entry point for all actions in Spark.
- rdd
target RDD to run tasks on
- func
a function to run on each partition of the RDD
- partitions
set of partitions to run on; some jobs may not want to compute on all partitions of the target RDD, e.g. for operations like
first()
- resultHandler
callback to pass each result to
-
def
sequenceFile[K, V](path: String, minPartitions: Int = defaultMinPartitions)(implicit km: ClassTag[K], vm: ClassTag[V], kcf: () ⇒ WritableConverter[K], vcf: () ⇒ WritableConverter[V]): RDD[(K, V)]
Version of sequenceFile() for types implicitly convertible to Writables through a WritableConverter.
Version of sequenceFile() for types implicitly convertible to Writables through a WritableConverter. For example, to access a SequenceFile where the keys are Text and the values are IntWritable, you could simply write
sparkContext.sequenceFile[String, Int](path, ...)
WritableConverters are provided in a somewhat strange way (by an implicit function) to support both subclasses of Writable and types for which we define a converter (e.g. Int to IntWritable). The most natural thing would've been to have implicit objects for the converters, but then we couldn't have an object for every subclass of Writable (you can't have a parameterized singleton object). We use functions instead to create a new converter for the appropriate type. In addition, we pass the converter a ClassTag of its type to allow it to figure out the Writable class to use in the subclass case.
- path
directory to the input data files, the path can be comma separated paths as a list of inputs
- minPartitions
suggested minimum number of partitions for the resulting RDD
- returns
RDD of tuples of key and corresponding value
- Note
Because Hadoop's RecordReader class re-uses the same Writable object for each record, directly caching the returned RDD or directly passing it to an aggregation or shuffle operation will create many references to the same object. If you plan to directly cache, sort, or aggregate Hadoop writable objects, you should first copy them using a
map
function.
-
def
sequenceFile[K, V](path: String, keyClass: Class[K], valueClass: Class[V]): RDD[(K, V)]
Get an RDD for a Hadoop SequenceFile with given key and value types.
Get an RDD for a Hadoop SequenceFile with given key and value types.
- path
directory to the input data files, the path can be comma separated paths as a list of inputs
- keyClass
Class
of the key associated withSequenceFileInputFormat
- valueClass
Class
of the value associated withSequenceFileInputFormat
- returns
RDD of tuples of key and corresponding value
- Note
Because Hadoop's RecordReader class re-uses the same Writable object for each record, directly caching the returned RDD or directly passing it to an aggregation or shuffle operation will create many references to the same object. If you plan to directly cache, sort, or aggregate Hadoop writable objects, you should first copy them using a
map
function.
-
def
sequenceFile[K, V](path: String, keyClass: Class[K], valueClass: Class[V], minPartitions: Int): RDD[(K, V)]
Get an RDD for a Hadoop SequenceFile with given key and value types.
Get an RDD for a Hadoop SequenceFile with given key and value types.
- path
directory to the input data files, the path can be comma separated paths as a list of inputs
- keyClass
Class
of the key associated withSequenceFileInputFormat
- valueClass
Class
of the value associated withSequenceFileInputFormat
- minPartitions
suggested minimum number of partitions for the resulting RDD
- returns
RDD of tuples of key and corresponding value
- Note
Because Hadoop's RecordReader class re-uses the same Writable object for each record, directly caching the returned RDD or directly passing it to an aggregation or shuffle operation will create many references to the same object. If you plan to directly cache, sort, or aggregate Hadoop writable objects, you should first copy them using a
map
function.
-
def
setCallSite(shortCallSite: String): Unit
Set the thread-local property for overriding the call sites of actions and RDDs.
-
def
setCheckpointDir(directory: String): Unit
Set the directory under which RDDs are going to be checkpointed.
Set the directory under which RDDs are going to be checkpointed.
- directory
path to the directory where checkpoint files will be stored (must be HDFS path if running in cluster)
-
def
setInterruptOnCancel(interruptOnCancel: Boolean): Unit
Set the behavior of job cancellation from jobs started in this thread.
Set the behavior of job cancellation from jobs started in this thread.
- interruptOnCancel
If true, then job cancellation will result in
Thread.interrupt()
being called on the job's executor threads. This is useful to help ensure that the tasks are actually stopped in a timely manner, but is off by default due to HDFS-1208, where HDFS may respond to Thread.interrupt() by marking nodes as dead.
- Since
3.5.0
-
def
setJobDescription(value: String): Unit
Set a human readable description of the current job.
-
def
setJobGroup(groupId: String, description: String, interruptOnCancel: Boolean = false): Unit
Assigns a group ID to all the jobs started by this thread until the group ID is set to a different value or cleared.
Assigns a group ID to all the jobs started by this thread until the group ID is set to a different value or cleared.
Often, a unit of execution in an application consists of multiple Spark actions or jobs. Application programmers can use this method to group all those jobs together and give a group description. Once set, the Spark web UI will associate such jobs with this group.
The application can also use
org.apache.spark.SparkContext.cancelJobGroup
to cancel all running jobs in this group. For example,// In the main thread: sc.setJobGroup("some_job_to_cancel", "some job description") sc.parallelize(1 to 10000, 2).map { i => Thread.sleep(10); i }.count() // In a separate thread: sc.cancelJobGroup("some_job_to_cancel")
- interruptOnCancel
If true, then job cancellation will result in
Thread.interrupt()
being called on the job's executor threads. This is useful to help ensure that the tasks are actually stopped in a timely manner, but is off by default due to HDFS-1208, where HDFS may respond to Thread.interrupt() by marking nodes as dead.
-
def
setLocalProperty(key: String, value: String): Unit
Set a local property that affects jobs submitted from this thread, such as the Spark fair scheduler pool.
Set a local property that affects jobs submitted from this thread, such as the Spark fair scheduler pool. User-defined properties may also be set here. These properties are propagated through to worker tasks and can be accessed there via org.apache.spark.TaskContext#getLocalProperty.
These properties are inherited by child threads spawned from this thread. This may have unexpected consequences when working with thread pools. The standard java implementation of thread pools have worker threads spawn other worker threads. As a result, local properties may propagate unpredictably.
-
def
setLogLevel(logLevel: String): Unit
Control our logLevel.
Control our logLevel. This overrides any user-defined log settings.
- logLevel
The desired log level as a string. Valid log levels include: ALL, DEBUG, ERROR, FATAL, INFO, OFF, TRACE, WARN
- val sparkUser: String
- val startTime: Long
- def statusTracker: SparkStatusTracker
-
def
stop(exitCode: Int): Unit
Shut down the SparkContext with exit code that will passed to scheduler backend.
Shut down the SparkContext with exit code that will passed to scheduler backend. In client mode, client side may call
SparkContext.stop()
to clean up but exit with code not equal to 0. This behavior cause resource scheduler such asApplicationMaster
exit with success status but client side exited with failed status. Spark can call this method to stop SparkContext and pass client side correct exit code to scheduler backend. Then scheduler backend should send the exit code to corresponding resource scheduler to keep consistent.- exitCode
Specified exit code that will passed to scheduler backend in client mode.
-
def
stop(): Unit
Shut down the SparkContext.
-
def
submitJob[T, U, R](rdd: RDD[T], processPartition: (Iterator[T]) ⇒ U, partitions: Seq[Int], resultHandler: (Int, U) ⇒ Unit, resultFunc: ⇒ R): SimpleFutureAction[R]
Submit a job for execution and return a FutureJob holding the result.
Submit a job for execution and return a FutureJob holding the result.
- rdd
target RDD to run tasks on
- processPartition
a function to run on each partition of the RDD
- partitions
set of partitions to run on; some jobs may not want to compute on all partitions of the target RDD, e.g. for operations like
first()
- resultHandler
callback to pass each result to
- resultFunc
function to be executed when the result is ready
-
final
def
synchronized[T0](arg0: ⇒ T0): T0
- Definition Classes
- AnyRef
-
def
textFile(path: String, minPartitions: Int = defaultMinPartitions): RDD[String]
Read a text file from HDFS, a local file system (available on all nodes), or any Hadoop-supported file system URI, and return it as an RDD of Strings.
Read a text file from HDFS, a local file system (available on all nodes), or any Hadoop-supported file system URI, and return it as an RDD of Strings. The text files must be encoded as UTF-8.
- path
path to the text file on a supported file system
- minPartitions
suggested minimum number of partitions for the resulting RDD
- returns
RDD of lines of the text file
-
def
toString(): String
- Definition Classes
- AnyRef → Any
- def uiWebUrl: Option[String]
-
def
union[T](first: RDD[T], rest: RDD[T]*)(implicit arg0: ClassTag[T]): RDD[T]
Build the union of a list of RDDs passed as variable-length arguments.
-
def
union[T](rdds: Seq[RDD[T]])(implicit arg0: ClassTag[T]): RDD[T]
Build the union of a list of RDDs.
-
def
version: String
The version of Spark on which this application is running.
-
final
def
wait(): Unit
- Definition Classes
- AnyRef
- Annotations
- @throws( ... )
-
final
def
wait(arg0: Long, arg1: Int): Unit
- Definition Classes
- AnyRef
- Annotations
- @throws( ... )
-
final
def
wait(arg0: Long): Unit
- Definition Classes
- AnyRef
- Annotations
- @throws( ... ) @native()
-
def
wholeTextFiles(path: String, minPartitions: Int = defaultMinPartitions): RDD[(String, String)]
Read a directory of text files from HDFS, a local file system (available on all nodes), or any Hadoop-supported file system URI.
Read a directory of text files from HDFS, a local file system (available on all nodes), or any Hadoop-supported file system URI. Each file is read as a single record and returned in a key-value pair, where the key is the path of each file, the value is the content of each file. The text files must be encoded as UTF-8.
For example, if you have the following files:
hdfs://a-hdfs-path/part-00000 hdfs://a-hdfs-path/part-00001 ... hdfs://a-hdfs-path/part-nnnnn
Do
val rdd = sparkContext.wholeTextFile("hdfs://a-hdfs-path")
,then
rdd
contains(a-hdfs-path/part-00000, its content) (a-hdfs-path/part-00001, its content) ... (a-hdfs-path/part-nnnnn, its content)
- path
Directory to the input data files, the path can be comma separated paths as the list of inputs.
- minPartitions
A suggestion value of the minimal splitting number for input data.
- returns
RDD representing tuples of file path and the corresponding file content
- Note
Small files are preferred, large file is also allowable, but may cause bad performance.
,On some filesystems,
,.../path/*
can be a more efficient way to read all files in a directory rather than.../path/
or.../path
Partitioning is determined by data locality. This may result in too few partitions by default.