ODIn/spark-instrumented-optimizer
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Implement PairRDDFunctions.partitionBy

Browse source
This commit is contained in:
Ankur Dave 2011-10-09 15:52:09 -07:00
parent 06637cb69e
commit 2d7057bf5d
1 changed files with 9 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
core/src/main/scala/spark/PairRDDFunctions.scala
View file

@ -77,6 +77,15 @@ class PairRDDFunctions[K: ClassManifest, V: ClassManifest](self: RDD[(K, V)]) ex
bufs.asInstanceOf[RDD[(K, Seq[V])]] bufs.asInstanceOf[RDD[(K, Seq[V])]]
} }
def partitionBy(partitioner: Partitioner): RDD[(K, V)] = {
def createCombiner(v: V) = ArrayBuffer(v)
def mergeValue(buf: ArrayBuffer[V], v: V) = buf += v
def mergeCombiners(b1: ArrayBuffer[V], b2: ArrayBuffer[V]) = b1 ++= b2
val bufs = combineByKey[ArrayBuffer[V]](
createCombiner _, mergeValue _, mergeCombiners _, defaultParallelism, partitioner)
bufs.flatMapValues(buf => buf)
}
def join[W](other: RDD[(K, W)], numSplits: Int): RDD[(K, (V, W))] = { def join[W](other: RDD[(K, W)], numSplits: Int): RDD[(K, (V, W))] = {
val vs: RDD[(K, Either[V, W])] = self.map { case (k, v) => (k, Left(v)) } val vs: RDD[(K, Either[V, W])] = self.map { case (k, v) => (k, Left(v)) }
val ws: RDD[(K, Either[V, W])] = other.map { case (k, w) => (k, Right(w)) } val ws: RDD[(K, Either[V, W])] = other.map { case (k, w) => (k, Right(w)) }