Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

IndexedRDD for accelerated joins #848

Open
wants to merge 17 commits into
base: master
Choose a base branch
from
Open

IndexedRDD for accelerated joins #848

Show file tree
Hide file tree
Changes from 11 commits
Commits
Show all changes
17 commits
Select commit Hold shift + click to select a range
f2b8dd3
second indexedrdd design
jegonzal Aug 13, 2013
b71d4fe
Finished early prototype of IndexedRDD
jegonzal Aug 14, 2013
54b5490
Adding testing code for indexedrdd
jegonzal Aug 14, 2013
6128175
IndexedRDD passes all PairRDD Function tests
jegonzal Aug 15, 2013
3bb6e01
adding better error handling when indexing an RDD
jegonzal Aug 15, 2013
327a4db
changing caching behavior on indexedrdds
jegonzal Aug 15, 2013
2b56852
Merge branch 'master' of https://github.com/mesos/spark into indexed_…
jegonzal Aug 18, 2013
8fd37ad
Merged with changes to zipPartitions
jegonzal Aug 18, 2013
630281b
Corrected all indexed RDD tests.
jegonzal Aug 19, 2013
0598c10
Merge branch 'master' of https://github.com/mesos/spark into indexed_rdd
jegonzal Aug 19, 2013
023702c
Updating documentation.
jegonzal Aug 19, 2013
55c6e73
Wrapping the index in an opaque type.
jegonzal Aug 20, 2013
93503a7
Allowing RDD to select its implementation of PairRDDFunctions
jegonzal Aug 28, 2013
9afd0e2
Merging upstream changes.
jegonzal Aug 28, 2013
766b6fd
Fixing IndexedRDD unit tests.
jegonzal Aug 28, 2013
3325083
Removing commented out test code
jegonzal Aug 28, 2013
c0b392a
Mering changes from master
jegonzal Sep 18, 2013
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 11 additions & 0 deletions core/src/main/scala/spark/PairRDDFunctions.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -698,6 +698,17 @@ class PairRDDFunctions[K: ClassManifest, V: ClassManifest](
*/
def values: RDD[V] = self.map(_._2)


def indexed(numPartitions: Int): IndexedRDD[K,V] =
IndexedRDD(self.partitionBy(new HashPartitioner(numPartitions)))

def indexed(partitioner: Partitioner): IndexedRDD[K,V] =
IndexedRDD(self.partitionBy(partitioner))


def indexed(existingIndex: RDD[BlockIndex[K]] = null): IndexedRDD[K,V] =
IndexedRDD(self, existingIndex)

private[spark] def getKeyClass() = implicitly[ClassManifest[K]].erasure

private[spark] def getValueClass() = implicitly[ClassManifest[V]].erasure
Expand Down
9 changes: 9 additions & 0 deletions core/src/main/scala/spark/RDD.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,9 @@ import spark.rdd.ZippedRDD
import spark.rdd.ZippedPartitionsRDD2
import spark.rdd.ZippedPartitionsRDD3
import spark.rdd.ZippedPartitionsRDD4

import spark.rdd.{IndexedRDD, BlockIndex}

import spark.storage.StorageLevel
import spark.util.BoundedPriorityQueue

Expand Down Expand Up @@ -770,6 +773,12 @@ abstract class RDD[T: ClassManifest](
return buf.toArray
}


def makeIndex(partitioner: Option[Partitioner] = None): RDD[BlockIndex[T]] =
IndexedRDD.makeIndex(this, partitioner)



/**
* Return the first element in this RDD.
*/
Expand Down
Loading