apache/spark — pull request #1297 by ankurdave (2014-07-04 02:56:43+00:00)
Title: Add IndexedRDD, an efficient updatable key-value store

This PR adds IndexedRDD, an efficient key-value store built on RDDs. IndexedRDD extends `RDD[(Long, V)]` by enforcing key uniqueness and pre-indexing the entries for efficient joins and point lookups, updates, and deletions.

It is implemented by (1) hash-partitioning the entries by key, (2) maintaining a hash index within each partition, and (3) using purely functional (immutable and efficiently updatable) data structures to enable efficient modifications and deletions.

As a result, this PR also introduces package-private implementations of three purely functional data structures: ImmutableVector, ImmutableLongOpenHashSet, and ImmutableBitSet.

GraphX will be the first user of IndexedRDD. This PR uses IndexedRDD to replace the current implementation of VertexRDD.

Here is a usage example for IndexedRDD:

```scala
import org.apache.spark.rdd.IndexedRDD

// Create an RDD of key-value pairs with Long keys
val rdd = sc.parallelize((1 to 1000000).map(x => (x.toLong, 0)))
// Construct an IndexedRDD from the pairs, hash-partitioning and indexing the entries
val indexed = IndexedRDD(rdd).cache()

// Perform some point updates. Note that the original IndexedRDD is not modified
val indexed2 = indexed.put(100L, 1).put(1234L, 10873).cache()

// Perform some point lookups
indexed2.get(1234L) // => Some(10873)
indexed.get(1234L) // => Some(0)

// Efficiently join derived IndexedRDD with original
val indexed3 = indexed.innerJoin(indexed2) { (id, a, b) => b }.filter(_._2 != 0)
indexed3.collect // => Array((100L, 1), (1234L, 10873))
```