The RDD API By Example
aggregate
The aggregate function allows the user to apply two different reduce functions to the RDD. The first reduce function is applied within each partition to reduce the data within each partition into a single result. The second reduce function is used to combine the different reduced results of all partitions together to arrive at one final result. The ability to have two separate reduce functions for intra partition versus across partition reducing adds a lot of flexibility. For example the first reduce function can be the max function and the second one can be the sum function. The user also specifies an initial value. Here are some important facts.
The initial value is applied at both levels of reduce. So both at the intra partition reduction and across partition reduction.
Both reduce functions have to be commutative and associative.
Do not assume any execution order for either partition computations or combining partitions.
Why would one want to use two input data types? Let us assume we do an archaeological site survey using a metal detector. While walking through the site we take GPS coordinates of important findings based on the output of the metal detector. Later, we intend to draw an image of a map that highlights these locations using the aggregate function. In this case the zeroValue could be an area map with no highlights. The possibly huge set of input data is stored as GPS coordinates across many partitions. seqOp (first reducer) could convert the GPS coordinates to map coordinates and put a marker on the map at the respective position. combOp (second reducer) will receive these highlights as partial maps and combine them into a single final output map.
Listing Variants
def aggregate[U: ClassTag](zeroValue: U)(seqOp: (U, T) => U, combOp: (U, U) => U): U
Examples 1
Examples 2
In contrast to the previous example, this example has the empty string at the beginning of the second partition. This results in length of zero being input to the second reduce which then upgrades it a length of 1. (Warning: The above example shows bad design since the output is dependent on the order of the data inside the partitions.)
aggregateByKey [Pair]
Works like the aggregate function except the aggregation is applied to the values with the same key. Also unlike the aggregate function the initial value is not applied to the second reduce.
def aggregateByKey[U](zeroValue: U)(seqOp: (U, V) ⇒ U, combOp: (U, U) ⇒ U)(implicit arg0: ClassTag[U]): RDD[(K, U)]
def aggregateByKey[U](zeroValue: U, numPartitions: Int)(seqOp: (U, V) ⇒ U, combOp: (U, U) ⇒ U)(implicit arg0: ClassTag[U]): RDD[(K, U)]
def aggregateByKey[U](zeroValue: U, partitioner: Partitioner)(seqOp: (U, V) ⇒ U, combOp: (U, U) ⇒ U)(implicit arg0: ClassTag[U]): RDD[(K, U)]
Example
cartesian
Computes the cartesian product between two RDDs (i.e. Each item of the first RDD is joined with each item of the second RDD) and returns them as a new RDD. (Warning: Be careful when using this function.! Memory consumption can quickly become an issue!)
def cartesian[U: ClassTag](other: RDD[U]): RDD[(T, U)]
checkpoint
Will create a checkpoint when the RDD is computed next. Checkpointed RDDs are stored as a binary file within the checkpoint directory which can be specified using the Spark context. (Warning: Spark applies lazy evaluation. Checkpointing will not occur until an action is invoked.)
Important note: the directory “my_directory_name” should exist in all slaves. As an alternative you could use an HDFS directory URL as well.
def checkpoint()
coalesce, repartition
Coalesces the associated data into a given number of partitions. repartition(numPartitions) is simply an abbreviation for coalesce(numPartitions, shuffle = true).
def coalesce ( numPartitions : Int , shuffle : Boolean = false ): RDD [T]
def repartition ( numPartitions : Int ): RDD [T]
cogroup [Pair], groupWith [Pair]
A very powerful set of functions that allow grouping up to 3 key-value RDDs together using their keys.
def cogroup[W](other: RDD[(K, W)]): RDD[(K, (Iterable[V], Iterable[W]))]
def cogroup[W](other: RDD[(K, W)], numPartitions: Int): RDD[(K, (Iterable[V], Iterable[W]))]
def cogroup[W](other: RDD[(K, W)], partitioner: Partitioner): RDD[(K, (Iterable[V], Iterable[W]))]
def cogroup[W1, W2](other1: RDD[(K, W1)], other2: RDD[(K, W2)]): RDD[(K, (Iterable[V], Iterable[W1], Iterable[W2]))]
def cogroup[W1, W2](other1: RDD[(K, W1)], other2: RDD[(K, W2)], numPartitions: Int): RDD[(K, (Iterable[V], Iterable[W1], Iterable[W2]))]
def cogroup[W1, W2](other1: RDD[(K, W1)], other2: RDD[(K, W2)], partitioner: Partitioner): RDD[(K, (Iterable[V], Iterable[W1], Iterable[W2]))]
def groupWith[W](other: RDD[(K, W)]): RDD[(K, (Iterable[V], Iterable[W]))]
def groupWith[W1, W2](other1: RDD[(K, W1)], other2: RDD[(K, W2)]): RDD[(K, (Iterable[V], IterableW1], Iterable[W2]))]
Examples
collect, toArray
Converts the RDD into a Scala array and returns it. If you provide a standard map-function (i.e. f = T -> U) it will be applied before inserting the values into the result array.
def collect(): Array[T]
def collect[U: ClassTag](f: PartialFunction[T, U]): RDD[U]
def toArray(): Array[T]
collectAsMap [Pair]
Similar to collect, but works on key-value RDDs and converts them into Scala maps to preserve their key-value structure.
def collectAsMap(): Map[K, V]
combineByKey[Pair]
Very efficient implementation that combines the values of a RDD consisting of two-component tuples by applying multiple aggregators one after another.
def combineByKey[C](createCombiner: V => C, mergeValue: (C, V) => C, mergeCombiners: (C, C) => C): RDD[(K, C)]
def combineByKey[C](createCombiner: V => C, mergeValue: (C, V) => C, mergeCombiners: (C, C) => C, numPartitions: Int): RDD[(K, C)]
def combineByKey[C](createCombiner: V => C, mergeValue: (C, V) => C, mergeCombiners: (C, C) => C, partitioner: Partitioner, mapSideCombine: Boolean = true, serializerClass: String = null): RDD[(K, C)]
compute
Executes dependencies and computes the actual representation of the RDD. This function should not be called directly by users.
def compute(split: Partition, context: TaskContext): Iterator[T]
context, sparkContext
Returns the SparkContext that was used to create the RDD.
count
Returns the number of items stored within a RDD.
def count(): Long
countApprox
Marked as experimental feature! Experimental features are currently not covered by this document!
def (timeout: Long, confidence: Double = 0.95): PartialResult[BoundedDouble]
countApproxDistinct
Computes the approximate number of distinct values. For large RDDs which are spread across many nodes, this function may execute faster than other counting methods. The parameter relativeSD controls the accuracy of the computation.
def countApproxDistinct(relativeSD: Double = 0.05): Long
countApproxDistinctByKey [Pair]
Similar to countApproxDistinct, but computes the approximate number of distinct values for each distinct key. Hence, the RDD must consist of two-component tuples. For large RDDs which are spread across many nodes, this function may execute faster than other counting methods. The parameter relativeSD controls the accuracy of the computation.
def countApproxDistinctByKey(relativeSD: Double = 0.05): RDD[(K, Long)]
def countApproxDistinctByKey(relativeSD: Double, numPartitions: Int): RDD[(K, Long)]
def countApproxDistinctByKey(relativeSD: Double, partitioner: Partitioner): RDD[(K, Long)]
val a = sc.parallelize(List(“Gnu”, “Cat”, “Rat”, “Dog”), 2)
val b = sc.parallelize(a.takeSample(true, 10000, 0), 20)
val c = sc.parallelize(1 to b.count().toInt, 20)
val d = b.zip(c)
d.countApproxDistinctByKey(0.1).collect
res15: Array[(String, Long)] = Array((Rat,2567), (Cat,3357), (Dog,2414), (Gnu,2494))
d.countApproxDistinctByKey(0.01).collect
res16: Array[(String, Long)] = Array((Rat,2555), (Cat,2455), (Dog,2425), (Gnu,2513))
d.countApproxDistinctByKey(0.001).collect
res0: Array[(String, Long)] = Array((Rat,2562), (Cat,2464), (Dog,2451), (Gnu,2521))
countByKey [Pair]
Very similar to count, but counts the values of a RDD consisting of two-component tuples for each distinct key separately.
def countByKey(): Map[K, Long]
Examval c = sc.parallelize(List((3, “Gnu”), (3, “Yak”), (5, “Mouse”), (3, “Dog”)), 2)
c.countByKey
res3: scala.collection.Map[Int,Long] = Map(3 -> 3, 5 -> 1)countByKeyApprox [Pair]
def countByKeyApprox(timeout: Long, confidence: Double = 0.95): PartialResult[Map[K, BoundedDouble]]
countByValue
Returns a map that contains all unique values of the RDD and their respective occurrence counts. (Warning: This operation will finally aggregate the information in a single reducer.)
def countByValue(): Map[T, Long]
countByValueApprox
def countByValueApprox(timeout: Long, confidence: Double = 0.95): PartialResult[Map[T, BoundedDouble]]
dependencies
Returns the RDD on which this RDD depends.
final def dependencies: Seq[Dependency[_]]
distinct
Returns a new RDD that contains each unique value only once.
def distinct(): RDD[T]
def distinct(numPartitions: Int): RDD[T]
first
Looks for the very first data item of the RDD and returns it.
def first(): T
filter
Evaluates a boolean function for each data item of the RDD and puts the items for which the function returned true into the resulting RDD.
def filter(f: T => Boolean): RDD[T]
When you provide a filter function, it must be able to handle all data items contained in the RDD. Scala provides so-called partial functions to deal with mixed data-types. (Tip: Partial functions are very useful if you have some data which may be bad and you do not want to handle but for the good data (matching data) you want to apply some kind of map function. The following article is good. It teaches you about partial functions in a very nice way and explains why case has to be used for partial functions: article)
Examples for mixed data without partial functions
t.
Examples for mixed data with partial functions
Be careful! The above code works because it only checks the type itself! If you use operations on this type, you have to explicitly declare what type you want instead of any. Otherwise the compiler does (apparently) not know what bytecode it should produce:
filterByRange [Ordered]
Returns an RDD containing only the items in the key range specified. From our testing, it appears this only works if your data is in key value pairs and it has already been sorted by key.
def filterByRange(lower: K, upper: K): RDD[P]
filterWith (deprecated)
This is an extended version of filter. It takes two function arguments. The first argument must conform to Int -> T and is executed once per partition. It will transform the partition index to type T. The second function looks like (U, T) -> Boolean. T is the transformed partition index and U are the data items from the RDD. Finally the function has to return either true or false (i.e. Apply the filter).
def filterWith[A: ClassTag](constructA: Int => A)(p: (T, A) => Boolean): RDD[T]
flatMap
Similar to map, but allows emitting more than one item in the map function.
def flatMap[U: ClassTag](f: T => TraversableOnce[U]): RDD[U]
flatMapValues
Very similar to mapValues, but collapses the inherent structure of the values during mapping.
def flatMapValues[U](f: V => TraversableOnce[U]): RDD[(K, U)]
flatMapWith (deprecated)
Similar to flatMap, but allows accessing the partition index or a derivative of the partition index from within the flatMap-function.
def flatMapWith[A: ClassTag, U: ClassTag](constructA: Int => A, preservesPartitioning: Boolean = false)(f: (T, A) => Seq[U]): RDD[U]
fold
Aggregates the values of each partition. The aggregation variable within each partition is initialized with zeroValue.
def fold(zeroValue: T)(op: (T, T) => T): T
foldByKey [Pair]
Very similar to fold, but performs the folding separately for each key of the RDD. This function is only available if the RDD consists of two-component tuples.
def foldByKey(zeroValue: V)(func: (V, V) => V): RDD[(K, V)]
def foldByKey(zeroValue: V, numPartitions: Int)(func: (V, V) => V): RDD[(K, V)]
def foldByKey(zeroValue: V, partitioner: Partitioner)(func: (V, V) => V): RDD[(K, V)]
foreach
Executes an parameterless function for each data item.
def foreach(f: T => Unit)
foreachPartition
Executes an parameterless function for each partition. Access to the data items contained in the partition is provided via the iterator argument.
def foreachPartition(f: Iterator[T] => Unit)
foreachWith (Deprecated)
def foreachWith[A: ClassTag](constructA: Int => A)(f: (T, A) => Unit)
fullOuterJoin [Pair]
Performs the full outer join between two paired RDDs.
def fullOuterJoin[W](other: RDD[(K, W)], numPartitions: Int): RDD[(K, (Option[V], Option[W]))]
def fullOuterJoin[W](other: RDD[(K, W)]): RDD[(K, (Option[V], Option[W]))]
def fullOuterJoin[W](other: RDD[(K, W)], partitioner: Partitioner): RDD[(K, (Option[V], Option[W]))]
generator, setGenerator
Allows setting a string that is attached to the end of the RDD’s name when printing the dependency graph.
getCheckpointFile
Returns the path to the checkpoint file or null if RDD has not yet been checkpointed.
def getCheckpointFile: Option[String]
preferredLocations
Returns the hosts which are preferred by this RDD. The actual preference of a specific host depends on various assumptions.
getStorageLevel
Retrieves the currently set storage level of the RDD. This can only be used to assign a new storage level if the RDD does not have a storage level set yet. The example below shows the error you will get, when you try to reassign the storage level.
def getStorageLevel
glom
Assembles an array that contains all elements of the partition and embeds it in an RDD. Each returned array contains the contents of one partition.
def glom(): RDD[Array[T]]
groupBy
def groupBy[K: ClassTag](f: T => K): RDD[(K, Iterable[T])]
def groupBy[K: ClassTag](f: T => K, numPartitions: Int): RDD[(K, Iterable[T])]
def groupBy[K: ClassTag](f: T => K, p: Partitioner): RDD[(K, Iterable[T])]
groupByKey [Pair]
Very similar to groupBy, but instead of supplying a function, the key-component of each pair will automatically be presented to the partitioner.
def groupByKey(): RDD[(K, Iterable[V])]
def groupByKey(numPartitions: Int): RDD[(K, Iterable[V])]
def groupByKey(partitioner: Partitioner): RDD[(K, Iterable[V])]
histogram [Double]
These functions take an RDD of doubles and create a histogram with either even spacing (the number of buckets equals to bucketCount) or arbitrary spacing based on custom bucket boundaries supplied by the user via an array of double values. The result type of both variants is slightly different, the first function will return a tuple consisting of two arrays. The first array contains the computed bucket boundary values and the second array contains the corresponding count of values (i.e. the histogram). The second variant of the function will just return the histogram as an array of integers.
def histogram(bucketCount: Int): Pair[Array[Double], Array[Long]]
def histogram(buckets: Array[Double], evenBuckets: Boolean = false): Array[Long]
Example with even spacing
id
Retrieves the ID which has been assigned to the RDD by its device context.
val id: Int
intersection
Returns the elements in the two RDDs which are the same.
def intersection(other: RDD[T], numPartitions: Int): RDD[T]
def intersection(other: RDD[T], partitioner: Partitioner)(implicit ord: Ordering[T] = null): RDD[T]
def intersection(other: RDD[T]): RDD[T]
isCheckpointed
Indicates whether the RDD has been checkpointed. The flag will only raise once the checkpoint has really been created.
def isCheckpointed: Boolean
iterator
Returns a compatible iterator object for a partition of this RDD. This function should never be called directly.
join [Pair]
Performs an inner join using two key-value RDDs. Please note that the keys must be generally comparable to make this work.
def join[W](other: RDD[(K, W)]): RDD[(K, (V, W))]
def join[W](other: RDD[(K, W)], numPartitions: Int): RDD[(K, (V, W))]
def join[W](other: RDD[(K, W)], partitioner: Partitioner): RDD[(K, (V, W))]
keyBy
Constructs two-component tuples (key-value pairs) by applying a function on each data item. The result of the function becomes the key and the original data item becomes the value of the newly created tuples.
def keyBy[K](f: T => K): RDD[(K, T)]
keys [Pair]
Extracts the keys from all contained tuples and returns them in a new RDD.
def keys: RDD[K]
leftOuterJoin [Pair]
Performs an left outer join using two key-value RDDs. Please note that the keys must be generally comparable to make this work correctly.
def leftOuterJoin[W](other: RDD[(K, W)]): RDD[(K, (V, Option[W]))]
def leftOuterJoin[W](other: RDD[(K, W)], numPartitions: Int): RDD[(K, (V, Option[W]))]
def leftOuterJoin[W](other: RDD[(K, W)], partitioner: Partitioner): RDD[(K, (V, Option[W]))]
lookup
Scans the RDD for all keys that match the provided value and returns their values as a Scala sequence.
def lookup(key: K): Seq[V]
map
Applies a transformation function on each item of the RDD and returns the result as a new RDD.
def map[U: ClassTag](f: T => U): RDD[U]
mapPartitions
This is a specialized map that is called only once for each partition. The entire content of the respective partitions is available as a sequential stream of values via the input argument (Iterarator[T]). The custom function must return yet another Iterator[U]. The combined result iterators are automatically converted into a new RDD. Please note, that the tuples (3,4) and (6,7) are missing from the following result due to the partitioning we chose.
def mapPartitions[U: ClassTag](f: Iterator[T] => Iterator[U], preservesPartitioning: Boolean = false): RDD[U]
Example 1
The above program can also be written using flatMap as follows.
Example 2 using flatmap
val x = sc.parallelize(1 to 10, 3)
x.flatMap(List.fill(scala.util.Random.nextInt(10))(_)).collect
res1: Array[Int] = Array(1, 2, 3, 3, 3, 4, 4, 4, 4, 4, 4, 4, 4, 4, 5, 5, 6, 6, 6, 6, 6, 6, 6, 6, 7, 7, 7, 8, 8, 8, 8, 8, 8, 8, 8, 9, 9, 9, 9, 9, 10, 10, 10, 10, 10, 10, 10, 10)
mapPartitionsWithContext (deprecated and developer API)
Similar to mapPartitions, but allows accessing information about the processing state within the mapper.
def mapPartitionsWithContext[U: ClassTag](f: (TaskContext, Iterator[T]) => Iterator[U], preservesPartitioning: Boolean = false): RDD[U]
mapPartitionsWithIndex
Similar to mapPartitions, but takes two parameters. The first parameter is the index of the partition and the second is an iterator through all the items within this partition. The output is an iterator containing the list of items after applying whatever transformation the function encodes.
def mapPartitionsWithIndex[U: ClassTag](f: (Int, Iterator[T]) => Iterator[U], preservesPartitioning: Boolean = false): RDD[U]
mapPartitionsWithSplit
This method has been marked as deprecated in the API. So, you should not use this method anymore. Deprecated methods will not be covered in this document.
mapValues [Pair]
Takes the values of a RDD that consists of two-component tuples, and applies the provided function to transform each value. Then, it forms new two-component tuples using the key and the transformed value and stores them in a new RDD.
def mapValues[U](f: V => U): RDD[(K, U)]
mapWith (deprecated)
This is an extended version of map. It takes two function arguments. The first argument must conform to Int -> T and is executed once per partition. It will map the partition index to some transformed partition index of type T. This is where it is nice to do some kind of initialization code once per partition. Like create a Random number generator object. The second function must conform to (U, T) -> U. T is the transformed partition index and U is a data item of the RDD. Finally the function has to return a transformed data item of type U.
def mapWith[A: ClassTag, U: ClassTag](constructA: Int => A, preservesPartitioning: Boolean = false)(f: (T, A) => U): RDD[U]
max
Returns the largest element in the RDD
def max()(implicit ord: Ordering[T]): T
mean [Double], meanApprox [Double]
Calls stats and extracts the mean component. The approximate version of the function can finish somewhat faster in some scenarios. However, it trades accuracy for speed.
def mean(): Double
def meanApprox(timeout: Long, confidence: Double = 0.95): PartialResult[BoundedDouble]
min
Returns the smallest element in the RDD
def min()(implicit ord: Ordering[T]): T
name, setName
Allows a RDD to be tagged with a custom name.
@transient var name: String
def setName(_name: String)
partitionBy [Pair]
Repartitions as key-value RDD using its keys. The partitioner implementation can be supplied as the first argument.
partitioner
Specifies a function pointer to the default partitioner that will be used for groupBy, subtract, reduceByKey (from PairedRDDFunctions), etc. functions.
partitions
Returns an array of the partition objects associated with this RDD.
final def partitions: Array[Partition]
persist, cache
These functions can be used to adjust the storage level of a RDD. When freeing up memory, Spark will use the storage level identifier to decide which partitions should be kept. The parameterless variants persist() and cache() are just abbreviations for persist(StorageLevel.MEMORY_ONLY). (Warning: Once the storage level has been changed, it cannot be changed again!)
def cache(): RDD[T]
def persist(): RDD[T]
def persist(newLevel: StorageLevel): RDD[T]
pipe
Takes the RDD data of each partition and sends it via stdin to a shell-command. The resulting output of the command is captured and returned as a RDD of string values.
def pipe(command: String): RDD[String]
def pipe(command: String, env: Map[String, String]): RDD[String]
def pipe(command: Seq[String], env: Map[String, String] = Map(), printPipeContext: (String => Unit) => Unit = null, printRDDElement: (T, String => Unit) => Unit = null): RDD[String]
v
randomSplit
Randomly splits an RDD into multiple smaller RDDs according to a weights Array which specifies the percentage of the total data elements that is assigned to each smaller RDD. Note the actual size of each smaller RDD is only approximately equal to the percentages specified by the weights Array. The second example below shows the number of items in each smaller RDD does not exactly match the weights Array. A random optional seed can be specified. This function is useful for spliting data into a training set and a testing set for machine learning.
def randomSplit(weights: Array[Double], seed: Long = Utils.random.nextLong): Array[RDD[T]]
reduce
This function provides the well-known reduce functionality in Spark. Please note that any function f you provide, should be commutative in order to generate reproducible results.
def reduce(f: (T, T) => T): T
reduceByKey [Pair], reduceByKeyLocally [Pair], reduceByKeyToDriver [Pair]
def reduceByKey(func: (V, V) => V): RDD[(K, V)]
def reduceByKey(func: (V, V) => V, numPartitions: Int): RDD[(K, V)]
def reduceByKey(partitioner: Partitioner, func: (V, V) => V): RDD[(K, V)]
def reduceByKeyLocally(func: (V, V) => V): Map[K, V]
def reduceByKeyToDriver(func: (V, V) => V): Map[K, V]
repartition
This function changes the number of partitions to the number specified by the numPartitions parameter
def repartition(numPartitions: Int)(implicit ord: Ordering[T] = null): RDD[T]
repartitionAndSortWithinPartitions [Ordered]
Repartition the RDD according to the given partitioner and, within each resulting partition, sort records by their keys.
def repartitionAndSortWithinPartitions(partitioner: Partitioner): RDD[(K, V)]
rightOuterJoin [Pair]
Performs an right outer join using two key-value RDDs. Please note that the keys must be generally comparable to make this work correctly.
def rightOuterJoin[W](other: RDD[(K, W)]): RDD[(K, (Option[V], W))]
def rightOuterJoin[W](other: RDD[(K, W)], numPartitions: Int): RDD[(K, (Option[V], W))]
def rightOuterJoin[W](other: RDD[(K, W)], partitioner: Partitioner): RDD[(K, (Option[V], W))]
sample
Randomly selects a fraction of the items of a RDD and returns them in a new RDD.
def sample(withReplacement: Boolean, fraction: Double, seed: Int): RDD[T]
sampleByKey [Pair]
Randomly samples the key value pair RDD according to the fraction of each key you want to appear in the final RDD.
def sampleByKey(withReplacement: Boolean, fractions: Map[K, Double], seed: Long = Utils.random.nextLong): RDD[(K, V)]
sampleByKeyExact [Pair, experimental]
This is labelled as experimental and so we do not document it.
saveAsHadoopFile [Pair], saveAsHadoopDataset [Pair], saveAsNewAPIHadoopFile [Pair]
Saves the RDD in a Hadoop compatible format using any Hadoop outputFormat class the user specifies.
def saveAsHadoopDataset(conf: JobConf)
def saveAsHadoopFile[F <: OutputFormat[K, V]](path: String)(implicit fm: ClassTag[F])
def saveAsHadoopFile[F <: OutputFormat[K, V]](path: String, codec: Class[_ <: CompressionCodec]) (implicit fm: ClassTag[F])
def saveAsHadoopFile(path: String, keyClass: Class[], valueClass: Class[], outputFormatClass: Class[_ <: OutputFormat[, ]], codec: Class[_ <: CompressionCodec])
def saveAsHadoopFile(path: String, keyClass: Class[], valueClass: Class[], outputFormatClass: Class[_ <: OutputFormat[, ]], conf: JobConf = new JobConf(self.context.hadoopConfiguration), codec: Option[Class[_ <: CompressionCodec]] = None)
def saveAsNewAPIHadoopFile[F <: NewOutputFormat[K, V]](path: String)(implicit fm: ClassTag[F])
def saveAsNewAPIHadoopFile(path: String, keyClass: Class[], valueClass: Class[], outputFormatClass: Class[_ <: NewOutputFormat[, ]], conf: Configuration = self.context.hadoopConfiguration)
saveAsObjectFile
Saves the RDD in binary format.
def saveAsObjectFile(path: String)
saveAsSequenceFile [SeqFile]
Saves the RDD as a Hadoop sequence file.
def saveAsSequenceFile(path: String, codec: Option[Class[_ <: CompressionCodec]] = None)
saveAsTextFile
Saves the RDD as text files. One line at a time.
def saveAsTextFile(path: String)
def saveAsTextFile(path: String, codec: Class[_ <: CompressionCodec])
Example without compression
stats [Double]
Simultaneously computes the mean, variance and the standard deviation of all values in the RDD.
def stats(): StatCounter
sortBy
This function sorts the input RDD’s data and stores it in a new RDD. The first parameter requires you to specify a function which maps the input data into the key that you want to sortBy. The second parameter (optional) specifies whether you want the data to be sorted in ascending or descending order.
def sortBy[K](f: (T) ⇒ K, ascending: Boolean = true, numPartitions: Int = this.partitions.size)(implicit ord: Ordering[K], ctag: ClassTag[K]): RDD[T]
sortByKey [Ordered]
This function sorts the input RDD’s data and stores it in a new RDD. The output RDD is a shuffled RDD because it stores data that is output by a reducer which has been shuffled. The implementation of this function is actually very clever. First, it uses a range partitioner to partition the data in ranges within the shuffled RDD. Then it sorts these ranges individually with mapPartitions using standard sort mechanisms.
def sortByKey(ascending: Boolean = true, numPartitions: Int = self.partitions.size): RDD[P]
stdev [Double], sampleStdev [Double]
Calls stats and extracts either stdev-component or corrected sampleStdev-component.
def stdev(): Double
def sampleStdev(): Double
subtract
Performs the well known standard set subtraction operation: A - B
def subtract(other: RDD[T]): RDD[T]
def subtract(other: RDD[T], numPartitions: Int): RDD[T]
def subtract(other: RDD[T], p: Partitioner): RDD[T]
subtractByKey [Pair]
Very similar to subtract, but instead of supplying a function, the key-component of each pair will be automatically used as criterion for removing items from the first RDD.
def subtractByKey[W: ClassTag](other: RDD[(K, W)]): RDD[(K, V)]
def subtractByKey[W: ClassTag](other: RDD[(K, W)], numPartitions: Int): RDD[(K, V)]
def subtractByKey[W: ClassTag](other: RDD[(K, W)], p: Partitioner): RDD[(K, V)]
sum [Double], sumApprox [Double]
Computes the sum of all values contained in the RDD. The approximate version of the function can finish somewhat faster in some scenarios. However, it trades accuracy for speed.
def sum(): Double
def sumApprox(timeout: Long, confidence: Double = 0.95): PartialResult[BoundedDouble]
take
Extracts the first n items of the RDD and returns them as an array. (Note: This sounds very easy, but it is actually quite a tricky problem for the implementors of Spark because the items in question can be in many different partitions.)
def take(num: Int): Array[T]
takeOrdered
Orders the data items of the RDD using their inherent implicit ordering function and returns the first n items as an array.
def takeOrdered(num: Int)(implicit ord: Ordering[T]): Array[T]
takeSample
Behaves different from sample in the following respects:
It will return an exact number of samples (Hint: 2nd parameter)
It returns an Array instead of RDD.
It internally randomizes the order of the items returned.
def takeSample(withReplacement: Boolean, num: Int, seed: Int): Array[T]
toDebugString
Returns a string that contains debug information about the RDD and its dependencies.
def toDebugString: String
toJavaRDD
Embeds this RDD object within a JavaRDD object and returns it.
def toJavaRDD() : JavaRDD[T]
toLocalIterator
Converts the RDD into a scala iterator at the master node.
def toLocalIterator: Iterator[T]
top
Utilizes the implicit ordering of T to determine the top k values and returns them as an array.
ddef top(num: Int)(implicit ord: Ordering[T]): Array[T]
toString
Assembles a human-readable textual description of the RDD.
override def toString: String
treeAggregate
Computes the same thing as aggregate, except it aggregates the elements of the RDD in a multi-level tree pattern. Another difference is that it does not use the initial value for the second reduce function (combOp). By default a tree of depth 2 is used, but this can be changed via the depth parameter.
def treeAggregate[U](zeroValue: U)(seqOp: (U, T) ⇒ U, combOp: (U, U) ⇒ U, depth: Int = 2)(implicit arg0: ClassTag[U]): U
treeReduce
Works like reduce except reduces the elements of the RDD in a multi-level tree pattern.
def treeReduce(f: (T, T) ⇒ T, depth: Int = 2): T
union, ++
Performs the standard set operation: A union B
def ++(other: RDD[T]): RDD[T]
def union(other: RDD[T]): RDD[T]
unpersist
Dematerializes the RDD (i.e. Erases all data items from hard-disk and memory). However, the RDD object remains. If it is referenced in a computation, Spark will regenerate it automatically using the stored dependency graph.
def unpersist(blocking: Boolean = true): RDD[T]
values
Extracts the values from all contained tuples and returns them in a new RDD.
def values: RDD[V]
variance [Double], sampleVariance [Double]
Calls stats and extracts either variance-component or corrected sampleVariance-component.
def variance(): Double
def sampleVariance(): Double
zip
Joins two RDDs by combining the i-th of either partition with each other. The resulting RDD will consist of two-component tuples which are interpreted as key-value pairs by the methods provided by the PairRDDFunctions extension.
def zip[U: ClassTag](other: RDD[U]): RDD[(T, U)]
zipParititions
Similar to zip. But provides more control over the zipping process.
def zipPartitions[B: ClassTag, V: ClassTag](rdd2: RDD[B])(f: (Iterator[T], Iterator[B]) => Iterator[V]): RDD[V]
def zipPartitions[B: ClassTag, V: ClassTag](rdd2: RDD[B], preservesPartitioning: Boolean)(f: (Iterator[T], Iterator[B]) => Iterator[V]): RDD[V]
def zipPartitions[B: ClassTag, C: ClassTag, V: ClassTag](rdd2: RDD[B], rdd3: RDD[C])(f: (Iterator[T], Iterator[B], Iterator[C]) => Iterator[V]): RDD[V]
def zipPartitions[B: ClassTag, C: ClassTag, V: ClassTag](rdd2: RDD[B], rdd3: RDD[C], preservesPartitioning: Boolean)(f: (Iterator[T], Iterator[B], Iterator[C]) => Iterator[V]): RDD[V]
def zipPartitions[B: ClassTag, C: ClassTag, D: ClassTag, V: ClassTag](rdd2: RDD[B], rdd3: RDD[C], rdd4: RDD[D])(f: (Iterator[T], Iterator[B], Iterator[C], Iterator[D]) => Iterator[V]): RDD[V]
def zipPartitions[B: ClassTag, C: ClassTag, D: ClassTag, V: ClassTag](rdd2: RDD[B], rdd3: RDD[C], rdd4: RDD[D], preservesPartitioning: Boolean)(f: (Iterator[T], Iterator[B], Iterator[C], Iterator[D]) => Iterator[V]): RDD[V]
zipWithIndex
Zips the elements of the RDD with its element indexes. The indexes start from 0. If the RDD is spread across multiple partitions then a spark Job is started to perform this operation.
def zipWithIndex(): RDD[(T, Long)]
zipWithUniqueId
This is different from zipWithIndex since just gives a unique id to each data element but the ids may not match the index number of the data element. This operation does not start a spark job even if the RDD is spread across multiple partitions.
Compare the results of the example below with that of the 2nd example of zipWithIndex. You should be able to see the difference.
def zipWithUniqueId(): RDD[(T, Long)]
刚开始使用SPARK的同学都会因为文档说明简单无示例而导致前期开发效率较低，在网上有一位老师的博客给出了很详细的使用示例，我简单将其翻译成中文，自己顺便也熟悉一下没使用过的API。
一些注解：
数据分片（partitions）：执行在计算节点中的一份数据集合，包含多个数据单元
以下为翻译内容：
RDD的API示例
RDD是弹性分布式数据集的简称，RDDs在Spark系统扮演干活的角色。当我们使用的时候，可以认为RDD是一个独立的数据分片集合，这些数据可以是计算后的数据结果。但是RDD实际上远不止这些，在集群中，独立的数据分片可以分布在不同的运算节点中。RDD是访问这些数据的入口，同时也能在这些数据上做运算和数据变换。不论RDD的数据丢失了或者部分丢失了，系统都能够通过lineage信息来恢复数据。Lineage是产生当前RDD的数据变化操作序列。总的来说，Spark可以从大部分的错误中恢复。
Spark中可以用的RDDs都是间接或者直接继承至RDD类。这个类包含很多处理数据分片的方法。RDD类是抽象类，平时大家用的都是RDD的继承实现类。继承类都需要实现一些核心方法，才能被外部使用。
Spark变成一个流行的大数据处理系统的原因是它没有对RDD分片中的数据增加限制。RDD API中包含了很多非常有用的方法。但是创作者为了保证核心API适配于所有的数据类型，所以有一些方便的方法并没有加到里面。
基础的RDD API将每个数据单元稻城单独的一个输入值。但是，使用者很多时候希望输入的是一个kV对，因此Spark提供了扩展的RDD类PairRDDFunctions。现在有四个RDD API扩展类。它们是：
DoubleRDDFunctions
这个扩展类包含了很多数值的聚合方法。如果RDD的数据单元能够隐式变换成Scala的double数据类型，这些方法会非常有用
PairRDDFunctions
该扩展类中的方法输入的数据单元是一个包含两个元素的tuple结构。Spark会把其中第一个元素当成key，第二个当成value。
OrderedRDDFunctions
该扩展类的方法需要输入数据是2元tuple，并且key能够排序。
SequenceFileRDDFunctions
这个扩展类包含一些可以创建Hadoop sequence文件的方法。输入数据必须是2元tuple。但需要额外考虑tuple元素能够转换成可写类型。
下面会按照字母顺序列出RDD方法，以下简称会在相应的API后面，说明该API属于哪个扩展类。
[Double] - DoubleRDDFunctions
[Ordered] - OrderedRDDFunctions
[Pair] - PairRDDFunctions
[SeqFile] - SequenceFileRDDFunctions
aggregate方法是一个高度定制化的RDD聚合和reduction方法，但是由于Scala和Spark在数据处理上的方式的原因，我们应该在真正使用的时候多加注意。下面是我们在使用中所观察到的要点：
reduce和combine方法是可代替的并且相互关联的
如方法定义中所述，combiner的输出必须是和输入类型一致，因为Spark是链式执行
zeroValue是U的初始值，作为 seqOp 和 combOp 执行之前的首个输入元素。你可以随着自己的需求改变它，但是为了代码执行结果的一致性，无论数据被分成多少份，每份有多大，都要保证产生同样的结果。
不要假设每个数据分片的执行顺序。
在每个数据分片上执行reduce之前，会设置一个zeroValue作为第一个输入，在执行combiner方法之前也会降zeroValue作为第一个输入。
为什么会有两个数据合并的输入函数？第一个函数将输入值映射到结果空间，输入值T可以和结果空间的值U不是一个类型，第二个函数将第一个函数映射后的结果再合并起来。
为什么有人需要输入两种数据类型的数据？此处举出一个示例，假设我们现在在考古遗址的现场，并且通过金属探测器找到重要的位置，并把位置的坐标记录下来。这样我们就可以通过聚合函数聚合这些标志出的坐标并画在地图上。我们可以设置 zeroValue 为一个空的地图。GPS坐标被存储在多个数据分片中， seqOp 可以将GPS坐标转化为地图坐标，并标注在 zeroValue 表示的地图上。 combOp 将这些标注在地图上的数据以及每个数据分片所代表的地图合并成一张完整的地图。
定义
相对于输入是数值，字符串输入会使人有一点点困惑，如果没想清楚aggregate功能的话。以示例2为例子，zeroValue为空字符串，假设第一个数据分片的输入是”12”,”23”，则第一次计算”“与”12”的最小长度是0，第二次计算以第一次计算结果做为输入”0”和”23”则最小长度为1，则此数据分片的最终输出为”1”，同理第二个数据分片也能得到输出”1”，最后combiner合并后可得”11”。
checkpoint在RDD被执行的时候调用。被checkpoint的RDD被存储在一个二进制文件中，文件存储在Spark context的初始设置的“checkpoint文件夹”里面。checkpoint文件夹必须存在于所有slave机器中，或者也可以设置成HDFS的地址。
将RDD数据重新分片，repartition是coalesce(numPartitions, shuffle = true)的缩写。
def coalesce ( numPartitions : Int , shuffle : Boolean = false ): RDD [T]def repartition ( numPartitions : Int ): RDD [T]
将需要cogroup的所有RDD对象放在一起，并将所有RDD中的所有tuple按key进行分组，得到的结果tuple包含当前tuple和其他tuple按可以分组后的value集合。具体参看示例。
将RDD转换为Scala数组。在collect中可以添加一个过滤函数对返回的数据进行初步处理，将原始数据T转化为目标数据U。
def collect(): Array[T]def collect[U: ClassTag](f: PartialFunction[T, U]): RDD[U]
类似collect方法，但是作用于key-value形态的RDDs，方法会返回Scala中map数据类型的kv结构数据。
在每个partition中先创建初始combiner（createCombiner），然后将当前RDD的数据单元逐个输入combiner进行处理（如拼装），每个partition处理后，再在mergeCombiner中将各个partition的结果进行综合处理。
返回SparkContext对象用来创建RDD。
返回RDD中数据单元的个数。
val c = sc.parallelize(List(“Gnu”, “Cat”, “Rat”, “Dog”), 2)
c.count
res2: Long = 4
针对二元组每个unique的key进行count。
返回unique的value所对应的个数。相对应countByKey。
快速计算count，但不精确。relativeSD控制精度。
针对RDD中的数据进行去重。
def distinct(): RDD[T]def distinct(numPartitions: Int): RDD[T]
返回RDD数据中第一个数据元素。
对RDD中的数据单元进行过滤。如果过滤函数f返回true，则当前被验证的数据单元会被过滤。
过滤函数需要和输入的数据类型匹配，如果输入数据是复合的数据类型可以利用partial functions。以下举了两个对比例子介绍。
以上是一个类型不匹配而导致失败的例子。
以下使用collect方法解决以上多类型数据的过滤问题，collect会使用isDefinedAt方法去判断当前输入是否符合，具体参见以下示例。
小心，上面的代码可行是因为它只是检测了数据的类型。如果你如果像下例中的情况，你需要将Any类型转换为你想要的类型。否则编译器不知道应该生成什么字节码：
val myfunc2: PartialFunction[Any, Any] = {case x if (x < 4) => “x”}
:10: error: value < is not a member of Any
val myfunc2: PartialFunction[Int, Any] = {case x if (x < 4) => “x”}
myfunc2: PartialFunction[Int,Any] =
这个方法是filter的扩展方法。它接收两组函数。第一组执行转换Int->T，这组在每个分片(partitions)上只执行一次。它会将分片中数据的index转化成T。第二组是(U,T)->Boolean。T是上一步index变换后的结果。U是RDD中的数据单元。最后这个函数返回ture或者false。
类似map方法，但是这个方法可以输出不止一个数据单元。
和mapValues很相似，此方法要求value是TraversableOnce类型的，输出结果会把TraversableOnce中的元素打散输出多个tuple出来。
聚合每个数据分片的数据。每个数据分片聚合之前可以设置一个初始值zeroValue。
功能类似fold，区别在于聚合范围不再是数据分别，而是每个key。输入的RDD数据必须是包含两个元素的tuple。
针对每个数据单元执行一个无参数的方法
针对每个数据分片执行当前输入的无参数方法，每个分片的数据元素通过迭代器访问。
获取checkpoint文件，如果RDD没有被checkpoint则返回null
将一个Array的数据分别分组和数据分片个数相同的Array中去，并生成相应RDD。
仅仅将数据分组，分组的key通过传入的函数生成。
针对数据类型为Double的RDD统计直方图。可以输入分桶数进行平均分桶，也可以输入自定义的分桶区间。平均分桶情况下会输出两个数组，第一个是每个分桶边界值，第二个是每个分桶的统计数。自定义分桶情况下只输出一个数组，即每个分桶的统计数。
def histogram(bucketCount: Int): Pair[Array[Double], Array[Long]]def histogram(buckets: Array[Double], evenBuckets: Boolean = false): Array[Long]
查询分配给RDD的id号
取两个RDD的交集
显示这个RDD是否被checkpoint了。谨记只有当执行了action操作之后相应的数据才会被checkpoint。
对两个key-value的RDD执行内连接。
针对两个key-value数据形式的RDD进行左外连接操作。
这是个特殊的map，它在每个数据分片上只执行一次。分片上的数据作为数据序列输入到处理函数，输入数据类型是迭代器(Iterarator[T])。处理函数返回另外一个迭代器。最终mapPartitions会调用合并程序将多个数据分片返回的数据合并成一个新的RDD。
功能和mapPartitions类似，但是有两个输入参数。第一个参数是数据分片的索引值，第二个参数是当前分片数据的迭代器。输出是当前分片数据处理后的数据迭代器。
va
将输入的二元tuple数据的value值逐一转换处理，并输出包含原有key和处理后value的二元tuple，最终输出处理后二元tuple的RDD。
返回RDD中最大的元素。
计算RDD中的均值，meanApprox提供近似算法，提高计算速度但是不精确。
def mean(): Doubledef meanApprox(timeout: Long, confidence: Double = 0.95): PartialResult[BoundedDouble]
返回RDD中最小的元素
给RDD打标签
@transient var name: Stringdef setName(_name: String)
将数据重新分片
def partitionBy(partitioner: Partitioner): RDD[(K, V)]
设置默认的partitioner，这个默认partitioner会在groupBy,subtract,reduceByKey等方法中用到。
@transient val partitioner: Option[Partitioner]
返回与当前RDD关联的partition对象。
这个函数可以用来调整RDD的存储等级。释放内存的时候，Spark需要根据这些存储等级标签决定释放谁。persist()和cache()是persist(StorageLevel.MEMORY_ONLY)的缩写。（Warning:一旦存储等级被改变，则它不能被再次改变！）
将RDD的每个数据分片都接到shell-command的标准输入上。经过shell-command的输出数据会重新生成新的RDD，新RDD是string类型的RDD。
数据集切分方法。随机将RDD的数据切分到多个小RDD中。切分比率由输入的权重Array确定。切分不会严格和输入的比率相同，同时也可以设定随机seed。
这个方法是Spark中很有名的方法，需要注意的是，其中输入的f的输入输出参数类型一致。
相对于reduce方法，这个方法会在同一key下进行reduce操作。reduceByKeyLocally与reduceByKey的区别是前者将数据返回到master节点中，并返回map结果，后者只返回RDD结果。
执行两个(key-value)RDD的右外连接。
随机取样
保存RDD为一个Hadoop的sequence文件
保存RDD为文本文件。一次一行
def saveAsTextFile(path: String)def saveAsTextFile(path: String, codec: Class[_ <: CompressionCodec])
这个方法将RDD的数据排序并存入新的RDD，第一个参数传入方法指定排序key，第二个参数指定是逆序还是顺序。
def sortBy[K](f: (T) �6�0 K, ascending: Boolean = true, numPartitions: Int = this.partitions.size)(implicit ord: Ordering[K], ctag: ClassTag[K]): RDD[T]
这个方法按Key进行排序，并输出新RDD。输出的RDD数据是经过shuffled之后的，因为在在输出的reducer里数据已经被shuffle了。
调用stats方法并抽取其中stdev结果或者sampleStdev结果。
def stdev(): Doubledef sampleStdev(): Double
实现差集
def sum(): Doubledef sumApprox(timeout: Long, confidence: Double = 0.95): PartialResult[BoundedDouble]
将RDD中前n个数据单元抽取出来作为array返回。
用RDD数据单元本身隐含的排序方法进行排序，排序后返回前n个元素组成的array。
以下是和sample的区别：
返回具体个数的样本（第二个参数指定）
直接返回array而不是RDD
内部会将返回结果随机打散
以string的形式返回RDD和它所依赖的RDD的调试信息
将RDD对象嵌入JavaRDD对象中并返回JavaRDD。
依着输入数据默认的排序机制，取出排序后的前n个value，并组成array返回。
将RDD组装出一个可读的文本。
执行集合合并操作
def ++(other: RDD[T]): RDD[T]def union(other: RDD[T]): RDD[T]
释放RDD，将存入磁盘和内存的数据元素释放。但是RDD对象会被保留，在后续使用它的时候，Spark会重新依据依赖图计算。
抽取所有tuple中的value并组装成新的RDD返回。
调用stats方法，并返回其中variance值或者sampleVariance值
def variance(): Doubledef sampleVariance(): Double
将两个RDD中第i个元素组成一个tuple，进而形成(key-value)的PairRDD。
功能与zip相近，但是可以对zip过程提供更多的控制。
将输入数据标号，标号从0开始。如果RDD有多个数据分片，则会启动一个Spark任务来处理。
与zipWithIndex不同的是这个方法会给每个数据单元一个独立的id，但是和数据单元的实际顺序的index无关。即使RDD存在多个数据分片，这个方法也不会启动spark任务去处理。
![手机软件抓包工具及其使用方法[图]](data:image/gif;base64,R0lGODlhAQABAIAAAP///wAAACwAAAAAAQABAAACAkQBADs=)