Github user andrewor14 commented on a diff in the pull request:
https://github.com/apache/spark/pull/9831#discussion_r45382352
--- Diff: core/src/test/scala/org/apache/spark/CheckpointSuite.scala ---
@@ -21,17 +21,222 @@ import java.io.File
import scala.reflect.ClassTag
+import org.apache.spark.CheckpointSuite._
import org.apache.spark.rdd._
import org.apache.spark.storage.{BlockId, StorageLevel, TestBlockId}
import org.apache.spark.util.Utils
+trait RDDCheckpointTester { self: SparkFunSuite =>
+
+ protected val partitioner = new HashPartitioner(2)
+
+ private def defaultCollectFunc[T](rdd: RDD[T]): Any = rdd.collect()
+
+ /** Implementations of this trait must implement this method */
+ protected def sparkContext: SparkContext
+
+ /**
+ * Test checkpointing of the RDD generated by the given operation. It
tests whether the
+ * serialized size of the RDD is reduce after checkpointing or not. This
function should be called
+ * on all RDDs that have a parent RDD (i.e., do not call on
ParallelCollection, BlockRDD, etc.).
+ *
+ * @param op an operation to run on the RDD
+ * @param reliableCheckpoint if true, use reliable checkpoints,
otherwise use local checkpoints
+ * @param collectFunc a function for collecting the values in the RDD,
in case there are
+ * non-comparable types like arrays that we want to
convert to something that supports ==
+ */
+ protected def testRDD[U: ClassTag](
+ op: (RDD[Int]) => RDD[U],
+ reliableCheckpoint: Boolean,
+ collectFunc: RDD[U] => Any = defaultCollectFunc[U] _): Unit = {
--- End diff --
indent 2 more
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
