From 9a54de16ed9de536e0436d532c587384e1ea0af6 Mon Sep 17 00:00:00 2001
From: Erik Erlandson <eerlands@redhat.com>
Date: Thu, 7 Aug 2014 23:45:16 -0700
Subject: [PATCH] [SPARK-2911]: provide rdd.parent[T](j) to obtain jth parent
RDD
Author: Erik Erlandson <eerlands@redhat.com>
Closes #1841 from erikerlandson/spark-2911-pr and squashes the following commits:
4699e2f [Erik Erlandson] [SPARK-2911]: provide rdd.parent[T](j) to obtain jth parent RDD
---
core/src/main/scala/org/apache/spark/rdd/RDD.scala | 5 +++++
.../src/test/scala/org/apache/spark/rdd/RDDSuite.scala | 10 ++++++++++
2 files changed, 15 insertions(+)
diff --git a/core/src/main/scala/org/apache/spark/rdd/RDD.scala b/core/src/main/scala/org/apache/spark/rdd/RDD.scala
index 0159003c88..19e10bd046 100644
--- a/core/src/main/scala/org/apache/spark/rdd/RDD.scala
+++ b/core/src/main/scala/org/apache/spark/rdd/RDD.scala
@@ -1233,6 +1233,11 @@ abstract class RDD[T: ClassTag](
dependencies.head.rdd.asInstanceOf[RDD[U]]
}
+ /** Returns the jth parent RDD: e.g. rdd.parent[T](0) is equivalent to rdd.firstParent[T] */
+ protected[spark] def parent[U: ClassTag](j: Int) = {
+ dependencies(j).rdd.asInstanceOf[RDD[U]]
+ }
+
/** The [[org.apache.spark.SparkContext]] that this RDD was created on. */
def context = sc
diff --git a/core/src/test/scala/org/apache/spark/rdd/RDDSuite.scala b/core/src/test/scala/org/apache/spark/rdd/RDDSuite.scala
index 4a7dc8dca2..926d4fecb5 100644
--- a/core/src/test/scala/org/apache/spark/rdd/RDDSuite.scala
+++ b/core/src/test/scala/org/apache/spark/rdd/RDDSuite.scala
@@ -726,6 +726,16 @@ class RDDSuite extends FunSuite with SharedSparkContext {
jrdd.rdd.retag.collect()
}
+ test("parent method") {
+ val rdd1 = sc.parallelize(1 to 10, 2)
+ val rdd2 = rdd1.filter(_ % 2 == 0)
+ val rdd3 = rdd2.map(_ + 1)
+ val rdd4 = new UnionRDD(sc, List(rdd1, rdd2, rdd3))
+ assert(rdd4.parent(0).isInstanceOf[ParallelCollectionRDD[_]])
+ assert(rdd4.parent(1).isInstanceOf[FilteredRDD[_]])
+ assert(rdd4.parent(2).isInstanceOf[MappedRDD[_, _]])
+ }
+
test("getNarrowAncestors") {
val rdd1 = sc.parallelize(1 to 100, 4)
val rdd2 = rdd1.filter(_ % 2 == 0).map(_ + 1)
--
GitLab