/**

* Applies a function f to all elements of this RDD.

*/

def foreach(f: T => Unit) {

val cleanF = sc.clean(f)

sc.runJob(this, (iter: Iterator[T]) => iter.foreach(cleanF))

}

/**

* Save this RDD as a text file, using string representations of elements.

*/

def saveAsTextFile(path: String) {

// https://issues.apache.org/jira/browse/SPARK-2075

//

// NullWritable is a `Comparable` in Hadoop 1.+, so the compiler cannot find an implicit

// Ordering for it and will use the default `null`. However, it's a `Comparable[NullWritable]`

// in Hadoop 2.+, so the compiler will call the implicit `Ordering.ordered` method to create an

// Ordering for `NullWritable`. That's why the compiler will generate different anonymous

// classes for `saveAsTextFile` in Hadoop 1.+ and Hadoop 2.+.

//

// Therefore, here we provide an explicit Ordering `null` to make sure the compiler generate

// same bytecodes for `saveAsTextFile`.

val nullWritableClassTag = implicitly[ClassTag[NullWritable]]

val textClassTag = implicitly[ClassTag[Text]]

val r = this.mapPartitions { iter =>

val text = new Text()

iter.map { x =>

text.set(x.toString)

(NullWritable.get(), text)

}

}

RDD.rddToPairRDDFunctions(r)(nullWritableClassTag, textClassTag, null)

.saveAsHadoopFile[TextOutputFormat[NullWritable, Text]](path)

}

/**

* Save this RDD as a compressed text file, using string representations of elements.

*/

def saveAsTextFile(path: String, codec: Class[_ <: CompressionCodec]) {

// https://issues.apache.org/jira/browse/SPARK-2075

val nullWritableClassTag = implicitly[ClassTag[NullWritable]]

val textClassTag = implicitly[ClassTag[Text]]

val r = this.mapPartitions { iter =>

val text = new Text()

iter.map { x =>

text.set(x.toString)

(NullWritable.get(), text)

}

}

RDD.rddToPairRDDFunctions(r)(nullWritableClassTag, textClassTag, null)

.saveAsHadoopFile[TextOutputFormat[NullWritable, Text]](path, codec)

}

/**

* Save this RDD as a SequenceFile of serialized objects.

*/

def saveAsObjectFile(path: String) {

this.mapPartitions(iter => iter.grouped(10).map(_.toArray))

.map(x => (NullWritable.get(), new BytesWritable(Utils.serialize(x))))

.saveAsSequenceFile(path)

}

/**

* Return an array that contains all of the elements in this RDD.

*/

def collect(): Array[T] = {

val results = sc.runJob(this, (iter: Iterator[T]) => iter.toArray)

Array.concat(results: _*)

}

/**

* Return the key-value pairs in this RDD to the master as a Map.

*

* Warning: this doesn't return a multimap (so if you have multiple values to the same key, only

* one value per key is preserved in the map returned)

*/

def collectAsMap(): Map[K, V] = {

val data = self.collect()

val map = new mutable.HashMap[K, V]

map.sizeHint(data.length)

data.foreach { pair => map.put(pair._1, pair._2) }

map

}

/**

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

*/

def reduceByKeyLocally(func: (V, V) => V): Map[K, V] = {

if (keyClass.isArray) {

throw new SparkException("reduceByKeyLocally() does not support array keys")

}

val reducePartition = (iter: Iterator[(K, V)]) => {

val map = new JHashMap[K, V]

iter.foreach { pair =>

val old = map.get(pair._1)

map.put(pair._1, if (old == null) pair._2 else func(old, pair._2))

}

Iterator(map)

} : Iterator[JHashMap[K, V]]

val mergeMaps = (m1: JHashMap[K, V], m2: JHashMap[K, V]) => {

m2.foreach { pair =>

val old = m1.get(pair._1)

m1.put(pair._1, if (old == null) pair._2 else func(old, pair._2))

}

m1

} : JHashMap[K, V]

self.mapPartitions(reducePartition).reduce(mergeMaps)

}

def lookup(key: K): Seq[V] = {

self.partitioner match {

case Some(p) =>

val index = p.getPartition(key)

val process = (it: Iterator[(K, V)]) => {

val buf = new ArrayBuffer[V]

for (pair <- it if pair._1 == key) {

buf += pair._2

}

buf

} : Seq[V]

val res = self.context.runJob(self, process, Array(index), false)

res(0)

case None =>

self.filter(_._1 == key).map(_._2).collect()

}

}

/**

* Return the number of elements in the RDD.

*/

def count(): Long = sc.runJob(this, Utils.getIteratorSize _).sum

/**

* 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(Seq(10, 4, 2, 12, 3)).top(1)

* // returns Array(12)

*

* sc.parallelize(Seq(2, 3, 4, 5, 6)).top(2)

* // returns Array(6, 5)

* }}}

*

* @param num k, the number of top elements to return

* @param ord the implicit ordering for T

* @return an array of top elements

*/

def top(num: Int)(implicit ord: Ordering[T]): Array[T] = takeOrdered(num)(ord.reverse)

/**

* Reduces the elements of this RDD using the specified commutative and

* associative binary operator.

*/

def reduce(f: (T, T) => T): T = {

val cleanF = sc.clean(f)

val reducePartition: Iterator[T] => Option[T] = iter => {

if (iter.hasNext) {

Some(iter.reduceLeft(cleanF))

} else {

None

}

}

var jobResult: Option[T] = None

val mergeResult = (index: Int, taskResult: Option[T]) => {

if (taskResult.isDefined) {

jobResult = jobResult match {

case Some(value) => Some(f(value, taskResult.get))

case None => taskResult

}

}

}

sc.runJob(this, reducePartition, mergeResult)

// Get the final result out of our Option, or throw an exception if the RDD was empty

jobResult.getOrElse(throw new UnsupportedOperationException("empty collection"))

}

/**

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

*/

def fold(zeroValue: T)(op: (T, T) => T): T = {

// Clone the zero value since we will also be serializing it as part of tasks

var jobResult = Utils.clone(zeroValue, sc.env.closureSerializer.newInstance())

val cleanOp = sc.clean(op)

val foldPartition = (iter: Iterator[T]) => iter.fold(zeroValue)(cleanOp)

val mergeResult = (index: Int, taskResult: T) => jobResult = op(jobResult, taskResult)

sc.runJob(this, foldPartition, mergeResult)

jobResult

}

/**

* Aggregate the elements of each partition, and then the results for all the partitions, using

* given combine functions and a neutral "zero value". This function can return a different result

* type, U, than the type of this RDD, T. Thus, we need one operation for merging a T into an U

* and one operation for merging two U's, as in scala.TraversableOnce. Both of these functions are

* allowed to modify and return their first argument instead of creating a new U to avoid memory

* allocation.

*/

def aggregate[U: ClassTag](zeroValue: U)(seqOp: (U, T) => U, combOp: (U, U) => U): U = {

// Clone the zero value since we will also be serializing it as part of tasks

var jobResult = Utils.clone(zeroValue, sc.env.closureSerializer.newInstance())

val cleanSeqOp = sc.clean(seqOp)

val cleanCombOp = sc.clean(combOp)

val aggregatePartition = (it: Iterator[T]) => it.aggregate(zeroValue)(cleanSeqOp, cleanCombOp)

val mergeResult = (index: Int, taskResult: U) => jobResult = combOp(jobResult, taskResult)

sc.runJob(this, aggregatePartition, mergeResult)

jobResult

}