RDD Operations
RDDs support two types of operations: transformations, which create a new dataset from an existing one, and actions, which return a value to the driver program after running a computation on the dataset. For example, map is a transformation that passes each dataset element through a function and returns a new RDD representing the results. On the other hand, reduce is an action that aggregates all the elements of the RDD using some function and returns the final result to the driver program (although there is also a parallel reduceByKey that returns a distributed dataset).
All transformations in Spark are lazy, in that they do not compute their results right away. Instead, they just remember the transformations applied to some base dataset (e.g. a file). The transformations are only computed when an action requires a result to be returned to the driver program. This design enables Spark to run more efficiently – for example, we can realize that a dataset created through map will be used in a reduce and return only the result of the reduce to the driver, rather than the larger mapped dataset.
@@僅有action需要結果時才會真正執行運算, 好處是可以提升執行的效率, 例如wordcount範例, faltMap與map產生的中間暫存結果就不需要傳回到user app
By default, each transformed RDD may be recomputed each time you run an action on it. However, you may also persist an RDD in memory using the persist (or cache) method, in which case Spark will keep the elements around on the cluster for much faster access the next time you query it. There is also support for persisting RDDs on disk, or replicated across multiple nodes.
@@可以用 persist 或 cache 函式保存tranformation過程的RDD內容
Basics
To illustrate RDD basics, consider the simple program below:
val lines = sc.textFile("data.txt")
val lineLengths = lines.map(s => s.length)
val totalLength = lineLengths.reduce((a, b) => a + b)
val lineLengths = lines.map(s => s.length)
val totalLength = lineLengths.reduce((a, b) => a + b)
The first line defines a base RDD from an external file. This dataset is not loaded in memory or otherwise acted on: lines is merely a pointer to the file. The second line defines lineLengths as the result of a map transformation. Again, lineLengths is not immediately computed, due to laziness. Finally, we run reduce, which is an action. At this point Spark breaks the computation into tasks to run on separate machines, and each machine runs both its part of the map and a local reduction, returning only its answer to the driver program.
@@textFile 僅建立指標指向外部檔案, map 僅記錄tramsformation過程, 執行 reduce 時才在各node執行map & reduce 運算task, 最後再將結果傳回user app
If we also wanted to use lineLengths again later, we could add:
lineLengths.persist()
before the reduce, which would cause lineLengths to be saved in memory after the first time it is computed.
Passing Functions to Spark
Spark’s API relies heavily on passing functions in the driver program to run on the cluster. There are two recommended ways to do this:
- Anonymous function syntax, which can be used for short pieces of code.
- Static methods in a global singleton object. For example, you can define object MyFunctions and then pass MyFunctions.func1, as follows:
object MyFunctions {
def func1(s: String): String = { ... }
}
myRdd.map(MyFunctions.func1)
def func1(s: String): String = { ... }
}
myRdd.map(MyFunctions.func1)
Note that while it is also possible to pass a reference to a method in a class instance (as opposed to a singleton object), this requires sending the object that contains that class along with the method. For example, consider:
class MyClass {
def func1(s: String): String = { ... }
def doStuff(rdd: RDD[String]): RDD[String] = { rdd.map(func1) }
}
def func1(s: String): String = { ... }
def doStuff(rdd: RDD[String]): RDD[String] = { rdd.map(func1) }
}
Here, if we create a new MyClass and call doStuff on it, the map inside there references the func1 method of that MyClass instance, so the whole object needs to be sent to the cluster. It is similar to writing rdd.map(x => this.func1(x)).
@@可以設計class的method處理RDD, 注意這裡用 RDD[String] 宣告傳入參數與傳回結果的型態, 也可以呼叫rdd.map傳入function literal呼叫class instance obj的method,
scala> val rdd1 = sc.textFile("hdfs://0a90e21c.cht.local:8020/tmp/s3log/S3log_2013012901_api1-8_cat")
scala> import org.apache.spark.rdd.RDD
scala> class MyClass {
def f1(s: String) = { s.split(" ").size.toString } def work1(rdd: RDD[String]): RDD[String] = { rdd.map(f1) } }
scala> val obj1 = new MyClass
scala> println(rdd1.first)
20130129000106, user1, user1, 10.144.30.153, 10.144.129.1, user1Bkt1, PutObject, 200, -, 10485760, -, 10485760
scala> println(obj1.work1(rdd1).first)
12
scala> println(rdd1.map(obj1.f1 _ ).first)
12
|
In a similar way, accessing fields of the outer object will reference the whole object:
class MyClass {
val field = "Hello"
def doStuff(rdd: RDD[String]): RDD[String] = { rdd.map(x => field + x) }
}
val field = "Hello"
def doStuff(rdd: RDD[String]): RDD[String] = { rdd.map(x => field + x) }
}
is equilvalent to writing rdd.map(x => this.field + x), which references all of this. To avoid this issue, the simplest way is to copy field into a local variable instead of accessing it externally:
def doStuff(rdd: RDD[String]): RDD[String] = {
val field_ = this.field
rdd.map(x => field_ + x)
}
@@可以設計class的method處理RDD時存取field, 也可以呼叫rdd.map傳入function literal存取class instance obj的field, 不過這樣需傳送整個obj至所有節點而影響效能, 可以將filed複製為變數給function literal使用
val field_ = this.field
rdd.map(x => field_ + x)
}
@@可以設計class的method處理RDD時存取field, 也可以呼叫rdd.map傳入function literal存取class instance obj的field, 不過這樣需傳送整個obj至所有節點而影響效能, 可以將filed複製為變數給function literal使用
scala> val rdd1 = sc.textFile("hdfs://0a90e21c.cht.local:8020/tmp/s3log/S3log_2013012901_api1-8_cat")
scala> import org.apache.spark.rdd.RDD
scala> class MyClass {
def f1(s: String) = { s.split(" ").size.toString } def work1(rdd: RDD[String]): RDD[String] = { rdd.map(f1) } val v1 = "Hi, " def f2(s:String) = { v1 + s.split(" ")(0) }
def work2(rdd: RDD[String]): RDD[String] = { rdd.map(f2) }
}
scala> val obj2 = new MyClass
scala> println(rdd1.first)
20130129000106, user1, user1, 10.144.30.153, 10.144.129.1, user1Bkt1, PutObject, 200, -, 10485760, -, 10485760
scala> println(obj2.work2(rdd1).first)
Hi, 20130129000106,
scala> println(rdd1.map(obj2.f2 _).first)
Hi, 20130129000106,
scala> println(rdd1.map(obj2.f2).first)
Hi, 20130129000106,
|
Working with Key-Value Pairs
While most Spark operations work on RDDs containing any type of objects, a few special operations are only available on RDDs of key-value pairs. The most common ones are distibuted “shuffle” operations, such as grouping or aggregating the elements by a key.
In Scala, these operations are automatically available on RDDs containing Tuple2 objects (the built-in tuples in the language, created by simply writing (a, b)), as long as you import org.apache.spark.SparkContext._ in your program to enable Spark’s implicit conversions. The key-value pair operations are available in the PairRDDFunctions class, which automatically wraps around an RDD of tuples if you import the conversions.
For example, the following code uses the reduceB yKey operation on key-value pairs to count how many times each line of text occurs in a file:
val lines = sc.textFile("data.txt")
val pairs = lines.map(s => (s, 1))
val counts = pairs.reduceByKey((a, b) => a + b)
val pairs = lines.map(s => (s, 1))
val counts = pairs.reduceByKey((a, b) => a + b)
scala> val rdd1 = sc.textFile("hdfs://0a90e21c.cht.local:8020/tmp/s3log/S3log_2013012901_api1-8_cat")
scala> val pairs = rdd1.flatMap(_.split(" ")).map((_ , 1))
pairs: org.apache.spark.rdd.RDD[(String, Int)] = MappedRDD[33] at map at <console>:17
scala> val counts = pairs.reduceByKey( _ + _ )
counts: org.apache.spark.rdd.RDD[(String, Int)] = MapPartitionsRDD[36] at reduceByKey at <console>:19
scala> val filters = counts.filter(_._1.contains("GetObject"))
filters: org.apache.spark.rdd.RDD[(String, Int)] = FilteredRDD[37] at filter at <console>:21
scala> filters.count
res40: Long = 1
@@由於RDD是分散儲存, 因此無法列印結果
scala> filters.foreach(kv => println("word["+ kv._1 +"] count["+ kv._2 +"]"))
...
14/06/11 17:42:23 INFO SparkContext: Job finished: foreach at <console>:24, took 0.117763755 s
scala> val res = filters.collect
res: Array[(String, Int)] = Array((GetObject,,6400000))
scala> filters.collect.foreach(kv => println("["+kv._1+"]["+kv._2+"]"))
[GetObject,][6400000]
scala>
|
We could also use counts.sortByKey(), for example, to sort the pairs alphabetically, and finally counts.collect() to bring them back to the driver program as an array of objects.
Transformations
The following table lists some of the common transformations supported by Spark. Refer to the RDD API doc (Scala, Java, Python) and pair RDD functions doc (Scala, Java) for details.
Meaning
| |||
|
Return the Cartesian product of this RDD and another one, that is, the RDD of all pairs of elements (a, b) where a is in this and b is in other.
When called on datasets of types T and U, returns a dataset of (T, U) pairs (all pairs of elements).
| |||
org.apache.spark.rdd.PairRDDFunctions
def cogroup[W1, W2](other1: RDD[(K, W1)], other2: RDD[(K, W2)], numPartitions: Int): RDD[(K, (Iterable[V], Iterable[W1],Iterable[W2]))]
For each key k in this or other1 or other2, return a resulting RDD that contains a tuple with the list of values for that key in this, other1 and other2.
For each key k in this or other, return a resulting RDD that contains a tuple with the list of values for that key in this as well as other.
def cogroup[W1, W2](other1: RDD[(K, W1)], other2: RDD[(K, W2)]): RDD[(K, (Iterable[V], Iterable[W1], Iterable[W2]))]
For each key k in this or other1 or other2, return a resulting RDD that contains a tuple with the list of values for that key in this, other1 and other2.
For each key k in this or other, return a resulting RDD that contains a tuple with the list of values for that key in this as well as other.
def cogroup[W1, W2](other1: RDD[(K, W1)], other2: RDD[(K, W2)], partitioner: Partitioner): RDD[(K, (Iterable[V], Iterable[W1],Iterable[W2]))]
For each key k in this or other1 or other2, return a resulting RDD that contains a tuple with the list of values for that key in this, other1 and other2.
For each key k in this or other, return a resulting RDD that contains a tuple with the list of values for that key in this as well as other.
When called on datasets of type (K, V) and (K, W), returns a dataset of (K, Iterable<V>, Iterable<W>) tuples. This operation is also called groupWith.
K, 1st RDD , 2nd RDD
| |||
def coalesce(numPartitions: Int, shuffle: Boolean = false)(implicit ord: Ordering[T] = null): RDD[T]
Return a new RDD that is reduced into numPartitions partitions.
This results in a narrow dependency, e.g. if you go from 1000 partitions to 100 partitions, there will not be a shuffle, instead each of the 100 new partitions will claim 10 of the current partitions.
However, if you're doing a drastic coalesce, e.g. to numPartitions = 1, this may result in your computation taking place on fewer nodes than you like (e.g. one node in the case of numPartitions = 1). To avoid this, you can pass shuffle = true. This will add a shuffle step, but means the current upstream partitions will be executed in parallel (per whatever the current partitioning is).
Note: With shuffle = true, you can actually coalesce to a larger number of partitions. This is useful if you have a small number of partitions, say 100, potentially with a few partitions being abnormally large. Calling coalesce(1000, shuffle = true) will result in 1000 partitions with the data distributed using a hash partitioner.
Decrease the number of partitions in the RDD to numPartitions. Useful for running operations more efficiently after filtering down a large dataset.
| |||
|
Return a new RDD containing the distinct elements in this RDD.
Return a new RDD containing the distinct elements in this RDD.
Return a new dataset that contains the distinct elements of the source dataset.
| |||
|
Return a new RDD containing only the elements that satisfy a predicate.
@@func傳入單一參數型態T, 傳回單一結果型態boolean
Return a new dataset formed by selecting those elements of the source on which func returns true.
| |||
|
Return a new RDD by first applying a function to all elements of this RDD, and then flattening the results.
@@func傳入單一參數型態T, 傳回多個結果Seq型態U
Similar to map, but each input item can be mapped to 0 or more output items (so func should return a Seq rather than a single item).
| |||
def groupBy[K](f: (T) ⇒ K, p: Partitioner)(implicit kt: ClassTag[K], ord: Ordering[K] = null): RDD[(K, Iterable[T])]
Return an RDD of grouped items. Each group consists of a key and a sequence of elements mapping to that key.
Return an RDD of grouped elements. Each group consists of a key and a sequence of elements mapping to that key.
Return an RDD of grouped items. Each group consists of a key and a sequence of elements mapping to that key.
org.apache.spark.rdd.PairRDDFunctions
Group the values for each key in the RDD into a single sequence. Hash-partitions the resulting RDD with the existing partitioner/parallelism level.
Note: If you are grouping in order to perform an aggregation (such as a sum or average) over each key, using PairRDDFunctions.reduceByKey or PairRDDFunctions.combineByKeywill provide much better performance,
Group the values for each key in the RDD into a single sequence. Hash-partitions the resulting RDD with into numPartitions partitions.
Note: If you are grouping in order to perform an aggregation (such as a sum or average) over each key, using PairRDDFunctions.reduceByKey or PairRDDFunctions.combineByKeywill provide much better performance.
Group the values for each key in the RDD into a single sequence. Allows controlling the partitioning of the resulting key-value pair RDD by passing a Partitioner.
Note: If you are grouping in order to perform an aggregation (such as a sum or average) over each key, using PairRDDFunctions.reduceByKey or PairRDDFunctions.combineByKeywill provide much better performance.
When called on a dataset of (K, V) pairs, returns a dataset of (K, Iterable<V>) pairs.
Note: If you are grouping in order to perform an aggregation (such as a sum or average) over each key, using reduceByKey or combineByKey will yield much better performance.
Note: By default, the level of parallelism in the output depends on the number of partitions of the parent RDD. You can pass an optional numTasks argument to set a different number of tasks.
| |||
org.apache.spark.rdd.DoubleRDDFunctions
def histogram(buckets: Array[Double], evenBuckets: Boolean = false): Array[Long]
Compute a histogram using the provided buckets. The buckets are all open to the left except for the last which is closed e.g. for the array [1, 10, 20, 50] the buckets are [1, 10) [10, 20) [20, 50] e.g 1<=x<10 , 10<=x<20, 20<=x<50 And on the input of 1 and 50 we would have a histogram of 1, 0, 0
Note: if your histogram is evenly spaced (e.g. [0, 10, 20, 30]) this can be switched from an O(log n) inseration to O(1) per element. (where n = # buckets) if you set evenBuckets to true. buckets must be sorted and not contain any duplicates. buckets array must be at least two elements All NaN entries are treated the same. If you have a NaN bucket it must be the maximum value of the last position and all NaN entries will be counted in that bucket.
def histogram(bucketCount: Int): (Array[Double], Array[Long])
Compute a histogram of the data using bucketCount number of buckets evenly spaced between the minimum and maximum of the RDD. For example if the min value is 0 and the max is 100 and there are two buckets the resulting buckets will be [0, 50) [50, 100]. bucketCount must be at least 1 If the RDD contains infinity, NaN throws an exception If the elements in RDD do not vary (max == min) always returns a single bucket.
| |||
|
Return the intersection of this RDD and another one. The output will not contain any duplicate elements, even if the input RDDs did. Note that this method performs a shuffle internally.
Return a new RDD that contains the intersection of elements in the source dataset and the argument.
| |||
org.apache.spark.rdd.PairRDDFunctions
Return an RDD containing all pairs of elements with matching keys in this and other. Each pair of elements will be returned as a (k, (v1, v2)) tuple, where (k, v1) is in this and (k, v2) is in other. Performs a hash join across the cluster.
Return an RDD containing all pairs of elements with matching keys in this and other. Each pair of elements will be returned as a (k, (v1, v2)) tuple, where (k, v1) is in this and (k, v2) is in other. Performs a hash join across the cluster.
Return an RDD containing all pairs of elements with matching keys in this and other. Each pair of elements will be returned as a (k, (v1, v2)) tuple, where (k, v1) is in this and (k, v2) is in other. Uses the given Partitioner to partition the output RDD.
When called on datasets of type (K, V) and (K, W), returns a dataset of (K, (V, W)) pairs with all pairs of elements for each key. Outer joins are also supported through leftOuterJoin and rightOuterJoin.
@@join產生tuple of (key, tuple of values)
Perform a left outer join of this and other. For each element (k, v) in this, the resulting RDD will either contain all pairs (k, (v, Some(w))) for w in other, or the pair (k, (v, None)) if no elements in other have key k. Hash-partitions the output into numPartitions partitions.
Perform a left outer join of this and other. For each element (k, v) in this, the resulting RDD will either contain all pairs (k, (v, Some(w))) for w in other, or the pair (k, (v, None)) if no elements in other have key k. Hash-partitions the output using the existing partitioner/parallelism level.
Perform a left outer join of this and other. For each element (k, v) in this, the resulting RDD will either contain all pairs (k, (v, Some(w))) for w in other, or the pair (k, (v, None)) if no elements in other have key k. Uses the given Partitioner to partition the output RDD.
Perform a right outer join of this and other. For each element (k, w) in other, the resulting RDD will either contain all pairs (k, (Some(v), w)) for v in this, or the pair (k, (None, w)) if no elements in this have key k. Hash-partitions the resulting RDD into the given number of partitions.
Perform a right outer join of this and other. For each element (k, w) in other, the resulting RDD will either contain all pairs (k, (Some(v), w)) for v in this, or the pair (k, (None, w)) if no elements in this have key k. Hash-partitions the resulting RDD using the existing partitioner/parallelism level.
Perform a right outer join of this and other. For each element (k, w) in other, the resulting RDD will either contain all pairs (k, (Some(v), w)) for v in this, or the pair (k, (None, w)) if no elements in this have key k. Uses the given Partitioner to partition the output RDD.
| |||
|
Return a new RDD by applying a function to all elements of this RDD.
@@func傳入單一參數型態T, 傳回單一結果型態T
Return a new distributed dataset formed by passing each element of the source through a function func.
| |||
def mapPartitions[U](f: (Iterator[T]) ⇒ Iterator[U], preservesPartitioning: Boolean = false)(implicit arg0: ClassTag[U]): RDD[U]
Return a new RDD by applying a function to each partition of this RDD.
@@func傳入多個參數Iterator型態T, 傳回多個結果Iterator型態U
Similar to map, but runs separately on each partition (block) of the RDD, so func must be of type Iterator<T> => Iterator<U> when running on an RDD of type T.
| |||
def mapPartitionsWithContext[U](f: (TaskContext, Iterator[T]) ⇒ Iterator[U], preservesPartitioning: Boolean = false)(implicit arg0: ClassTag[U]): RDD[U]
Return a new RDD by applying a function to each partition of this RDD. This is a variant of mapPartitions that also passes the TaskContext into the closure.
@@func傳入多個參數Iterator型態T, 傳回多個結果Iterator型態U
| |||
def mapPartitionsWithIndex[U](f: (Int, Iterator[T]) ⇒ Iterator[U], preservesPartitioning: Boolean = false)(implicit arg0: ClassTag[U]):RDD[U]
Return a new RDD by applying a function to each partition of this RDD, while tracking the index of the original partition.
@@func傳入多個參數Iterator型態T, 傳回多個結果Iterator型態U
Similar to mapPartitions, but also provides func with an integer value representing the index of the partition, so func must be of type (Int, Iterator<T>) => Iterator<U> when running on an RDD of type T.
| |||
def pipe(command: Seq[String], env: Map[String, String] = Map(), printPipeContext: ((String) ⇒ Unit) ⇒ Unit = null, printRDDElement: (T, (String) ⇒ Unit) ⇒ Unit = null, separateWorkingDir: Boolean = false): RDD[String]
Return an RDD created by piping elements to a forked external process. The print behavior can be customized by providing two functions.
command : command to run in forked process.
env : environment variables to set.
printPipeContext : Before piping elements, this function is called as an oppotunity to pipe context data. Print line function (like out.println) will be passed as printPipeContext's parameter.
printRDDElement : Use this function to customize how to pipe elements. This function will be called with each RDD element as the 1st parameter, and the print line function (like out.println()) as the 2nd parameter. An example of pipe the RDD data of groupBy() in a streaming way, instead of constructing a huge String to concat all the elements: def printRDDElement(record:(String, Seq[String]), f:String=>Unit) = for (e <- record._2){f(e)}
separateWorkingDir : Use separate working directories for each task.
returns : the result RDD
Return an RDD created by piping elements to a forked external process.
Return an RDD created by piping elements to a forked external process.
@@根據partition數目, 平行將RDD partition的成員導向command的stdin, 並將command的stdout收集為RDD[String]的partition
Pipe each partition of the RDD through a shell command, e.g. a Perl or bash script. RDD elements are written to the process's stdin and lines output to its stdout are returned as an RDD of strings.
| |||
org.apache.spark.rdd.PairRDDFunctions
Merge the values for each key using an associative reduce function. This will also perform the merging locally on each mapper before sending results to a reducer, similarly to a "combiner" in MapReduce. Output will be hash-partitioned with the existing partitioner/ parallelism level
Merge the values for each key using an associative reduce function. This will also perform the merging locally on each mapper before sending results to a reducer, similarly to a "combiner" in MapReduce. Output will be hash-partitioned with numPartitions partitions.
Merge the values for each key using an associative reduce function. This will also perform the merging locally on each mapper before sending results to a reducer, similarly to a "combiner" in MapReduce.
def reduceByKeyLocally(func: (V, V) ⇒ V): Map[K, V]
Merge the values for each key using an associative reduce function, but return the results immediately to the master as a Map. This will also perform the merging locally on each mapper before sending results to a reducer, similarly to a "combiner" in MapReduce.
When called on a dataset of (K, V) pairs, returns a dataset of (K, V) pairs where the values for each key are aggregated using the given reduce function. Like in groupByKey, the number of reduce tasks is configurable through an optional second argument.
| |||
|
Return a new RDD that has exactly numPartitions partitions.
Can increase or decrease the level of parallelism in this RDD. Internally, this uses a shuffle to redistribute data.
If you are decreasing the number of partitions in this RDD, consider using coalesce, which can avoid performing a shuffle.
@@可以增加或減少partition數目, 一定會產生shuffle步驟, 如果是減少partition的話可考慮使用colease不用shuffle
Reshuffle the data in the RDD randomly to create either more or fewer partitions and balance it across them. This always shuffles all data over the network.
| |||
|
Return a sampled subset of this RDD.
core/src/main/scala/org/apache/spark/rdd/RDD.scala
@@withR=true => use PoissonSampler, withR=false => BernoulliSampler
Sample a fraction fraction of the data, with or without replacement, using a given random number generator seed.
| |||
org.apache.spark.rdd.OrderedRDDFunctions
Sort the RDD by key, so that each partition contains a sorted range of the elements. Calling collect or save on the resulting RDD will return or output an ordered list of records (in thesave case, they will be written to multiple part-X files in the filesystem, in order of the keys).
When called on a dataset of (K, V) pairs where K implements Ordered, returns a dataset of (K, V) pairs sorted by keys in ascending or descending order, as specified in the boolean ascending argument.
| |||
|
Return the union of this RDD and another one.
Return a new dataset that contains the union of the elements in the source dataset and the argument.
|
@@對 RDD[key:String, value:Int] 計算每個Key的平均值
scala> val rdd1 = sc.parallelize( List(("k1",1),("k2",2),("k2",2),("k3",3),("k3",3),("k3",3)) )
rdd: org.apache.spark.rdd.RDD[(String, Int)] = ParallelCollectionRDD[8] at parallelize at <console>:12
scala> val rdd2 = rdd.groupByKey()
rdd2: org.apache.spark.rdd.RDD[(String, Iterable[Int])] = MappedValuesRDD[11] at groupByKey at <console>:14
scala> val rdd3 = rdd2.map( tup => (tup._1 , tup._2.sum/tup._2.size) )
scala> val col2 = rdd3.collect
col2: Array[(String, Int)] = Array((k1,1), (k2,2), (k3,3))
scala> val rdd1 = sc.parallelize( List(("k1",15),("k2",22),("k2",24),("k3",13),("k3",23),("k3",33)) )
scala> val col1 = rdd1.groupByKey().map( tup => (tup._1 , tup._2.sum / tup._2.size) ).collect
col1: Array[(String, Int)] = Array((k1,15), (k2,23), (k3,23))
|
Actions
The following table lists some of the common actions supported by Spark. Refer to the RDD API doc (Scala, Java, Python) and pair RDD functions doc (Scala, Java) for details.
def collect(): Array[T]
Return an array that contains all of the elements in this RDD.
Return all the elements of the dataset as an array at the driver program. This is usually useful after a filter or other operation that returns a sufficiently small subset of the data.
org.apache.spark.rdd.PairRDDFunctions
def collectAsMap(): Map[K, V]
Return the key-value pairs in this RDD to the master as a Map.
| ||
def count(): Long
Return the number of elements in the RDD.
Return the number of elements in the dataset.
| ||
org.apache.spark.rdd.PairRDDFunctions
def countByKey(): Map[K, Long]
Count the number of elements for each key, and return the result to the master as a Map.
Only available on RDDs of type (K, V). Returns a hashmap of (K, Int) pairs with the count of each key.
| ||
def first(): T
Return the first element in this RDD.
Return the first element of the dataset (similar to take(1)).
| ||
def fold(zeroValue: T)(op: (T, T) ⇒ T): T
Aggregate the elements of each partition, and then the results for all the partitions, using a given associative function and a neutral "zero value". The function op(t1, t2) is allowed to modify t1 and return it as its result value to avoid object allocation; however, it should not modify t2.
@@傳入函式op(t1, t2), 其中t1為上次運算結果result value, t2為RDD成員, 起始值zeroValue必須是RDD成員型態T
| ||
def foreach(f: (T) ⇒ Unit): Unit
Applies a function f to all elements of this RDD.
foreach(func)
Run a function func on each element of the dataset. This is usually done for side effects such as updating an accumulator variable (see below) or interacting with external storage systems.
| ||
def max()(implicit ord: Ordering[T]): T
Returns the max of this RDD as defined by the implicit Ordering[T].
| ||
org.apache.spark.rdd.DoubleRDDFunctions
def mean(): Double
Compute the mean of this RDD's elements.
Approximate operation to return the mean within a timeout.
| ||
def min()(implicit ord: Ordering[T]): T
Returns the min of this RDD as defined by the implicit Ordering[T].
| ||
def reduce(f: (T, T) ⇒ T): T
Reduces the elements of this RDD using the specified commutative and associative binary operator.
Aggregate the elements of the dataset using a function func (which takes two arguments and returns one). The function should be commutative and associative so that it can be computed correctly in parallel.
| ||
def saveAsObjectFile(path: String): Unit
Save this RDD as a SequenceFile of serialized objects.
Write the elements of the dataset in a simple format using Java serialization, which can then be loaded usingSparkContext.objectFile().
| ||
org.apache.spark.rdd.SequenceFileRDDFunctions
def saveAsSequenceFile(path: String, codec: Option[Class[_ <: CompressionCodec]] = None): Unit
Output the RDD as a Hadoop SequenceFile using the Writable types we infer from the RDD's key and value types. If the key or value are Writable, then we use their classes directly; otherwise we map primitive types such as Int and Double to IntWritable, DoubleWritable, etc, byte arrays to BytesWritable, and Strings to Text. The path can be on any Hadoop-supported file system.
Write the elements of the dataset as a Hadoop SequenceFile in a given path in the local filesystem, HDFS or any other Hadoop-supported file system. This is available on RDDs of key-value pairs that either implement Hadoop's Writable interface. In Scala, it is also available on types that are implicitly convertible to Writable (Spark includes conversions for basic types like Int, Double, String, etc).
| ||
def saveAsTextFile(path: String, codec: Class[_ <: CompressionCodec]): Unit
Save this RDD as a compressed text file, using string representations of elements.
def saveAsTextFile(path: String): Unit
Save this RDD as a text file, using string representations of elements.
Write the elements of the dataset as a text file (or set of text files) in a given directory in the local filesystem, HDFS or any other Hadoop-supported file system. Spark will call toString on each element to convert it to a line of text in the file.
| ||
|
Return a org.apache.spark.util.StatCounter object that captures the mean, variance and count of the RDD's elements in one operation.
@@參考sample()函式範例
| ||
def take(num: Int): Array[T]
Take the first num elements of the RDD. It works by first scanning one partition, and use the results from that partition to estimate the number of additional partitions needed to satisfy the limit.
Return an array with the first n elements of the dataset. Note that this is currently not executed in parallel. Instead, the driver program computes all the elements.
@@傳回Array, 成員型態為RDD的成員型態
| ||
def takeOrdered(num: Int)(implicit ord: Ordering[T]): Array[T]
Returns the first K (smallest) elements from this RDD as defined by the specified implicit Ordering[T] and maintains the ordering. This does the opposite of top. For example:
sc.parallelize([10, 4, 2, 12, 3]).takeOrdered(1)
// returns [12] sc.parallelize([2, 3, 4, 5, 6]).takeOrdered(2) // returns [2, 3]
num : the number of top elements to return
ord : the implicit ordering for T
returns : an array of top elements
Return the first n elements of the RDD using either their natural order or a custom comparator. @@排序後從最小值開始取num個成員, 傳回Array
| ||
|
Return an array with a random sample of num elements of the dataset, with or without replacement, using the given random number generator seed.
core/src/main/scala/org/apache/spark/rdd/RDD.scala
@@withR=true => use PoissonSampler, withR=false => BernoulliSampler
| ||
def top(num: Int)(implicit ord: Ordering[T]): Array[T]
Returns the top K (largest) elements from this RDD as defined by the specified implicit Ordering[T]. This does the opposite of takeOrdered. For example:
sc.parallelize([10, 4, 2, 12, 3]).top(1)
// returns [12] sc.parallelize([2, 3, 4, 5, 6]).top(2) // returns [6, 5]
num : the number of top elements to return
ord : the implicit ordering for T
returns : an array of top elements
|
