diff options
author | Matei Zaharia <matei@eecs.berkeley.edu> | 2013-01-05 19:15:54 -0800 |
---|---|---|
committer | Matei Zaharia <matei@eecs.berkeley.edu> | 2013-01-05 19:15:54 -0800 |
commit | b1663752c6cb6e5eea938cdff87e12ec05707e03 (patch) | |
tree | 639852a47f98a36a826a1f2f2038472d9d21b4aa | |
parent | 0982572519655354b10987de4f68e29b8331bd2a (diff) | |
parent | 6a0db3b449a829f3e5cdf7229f6ee564268be1df (diff) | |
download | spark-b1663752c6cb6e5eea938cdff87e12ec05707e03.tar.gz spark-b1663752c6cb6e5eea938cdff87e12ec05707e03.tar.bz2 spark-b1663752c6cb6e5eea938cdff87e12ec05707e03.zip |
Merge pull request #351 from stephenh/values
Add PairRDDFunctions.keys and values.
-rw-r--r-- | core/src/main/scala/spark/PairRDDFunctions.scala | 10 | ||||
-rw-r--r-- | core/src/test/scala/spark/ShuffleSuite.scala | 7 |
2 files changed, 17 insertions, 0 deletions
diff --git a/core/src/main/scala/spark/PairRDDFunctions.scala b/core/src/main/scala/spark/PairRDDFunctions.scala index 413c944a66..ce48cea903 100644 --- a/core/src/main/scala/spark/PairRDDFunctions.scala +++ b/core/src/main/scala/spark/PairRDDFunctions.scala @@ -615,6 +615,16 @@ class PairRDDFunctions[K: ClassManifest, V: ClassManifest]( writer.cleanup() } + /** + * Return an RDD with the keys of each tuple. + */ + def keys: RDD[K] = self.map(_._1) + + /** + * Return an RDD with the values of each tuple. + */ + def values: RDD[V] = self.map(_._2) + private[spark] def getKeyClass() = implicitly[ClassManifest[K]].erasure private[spark] def getValueClass() = implicitly[ClassManifest[V]].erasure diff --git a/core/src/test/scala/spark/ShuffleSuite.scala b/core/src/test/scala/spark/ShuffleSuite.scala index 8170100f1d..bebb8ebe86 100644 --- a/core/src/test/scala/spark/ShuffleSuite.scala +++ b/core/src/test/scala/spark/ShuffleSuite.scala @@ -216,6 +216,13 @@ class ShuffleSuite extends FunSuite with ShouldMatchers with BeforeAndAfter { // Test that a shuffle on the file works, because this used to be a bug assert(file.map(line => (line, 1)).reduceByKey(_ + _).collect().toList === Nil) } + + test("keys and values") { + sc = new SparkContext("local", "test") + val rdd = sc.parallelize(Array((1, "a"), (2, "b"))) + assert(rdd.keys.collect().toList === List(1, 2)) + assert(rdd.values.collect().toList === List("a", "b")) + } } object ShuffleSuite { |