Skip to content
Browse files

It compiles with the Java 8 happy pandas

  • Loading branch information...
1 parent e64e1ad commit 8bfd3f3e2c3721bf192c196ea1a3a2d4a2d63a32 @holdenk committed Mar 8, 2014
View
11 ...tion/MapPartitionsWithIndexFunction.scala → ...ction/MapPartitionsWithIndexFunction.java
@@ -15,16 +15,13 @@
* limitations under the License.
*/
-package org.apache.spark.api.java.function
+package org.apache.spark.api.java.function;
-import scala.reflect.ClassTag
-
-import org.apache.spark.api.java.JavaSparkContext.fakeClassTag
+import java.io.Serializable;
/**
* A function that takes the a partition
*/
-abstract class MapPartitionsWithIndexFunction[A, B] extends Function2[Integer,
- java.util.Iterator[A], java.util.Iterator[B]] {
- def elementType() : ClassTag[B] = fakeClassTag
+public interface MapPartitionsWithIndexFunction<A, B> extends Function2<Integer,
+ java.util.Iterator<A>, java.util.Iterator<B>> {
}
View
8 core/src/main/scala/org/apache/spark/api/java/JavaRDDLike.scala
@@ -73,12 +73,12 @@ trait JavaRDDLike[T, This <: JavaRDDLike[T, This]] extends Serializable {
* Return a new RDD by applying a function to each partition of this RDD, while tracking the index
* of the original partition.
*/
- def mapPartitionsWithIndex[R](f: MapPartitionsWithIndexFunction[T, R],
+ def mapPartitionsWithIndex[R: ClassTag](f: MapPartitionsWithIndexFunction[T, R],
preservesPartitioning: Boolean = false): JavaRDD[R] = {
import scala.collection.JavaConverters._
- def fn = (a: Int, b: Iterator[T]) => f.apply(a, asJavaIterator(b)).asScala
- val newRdd = rdd.mapPartitionsWithIndex(fn, preservesPartitioning)(f.elementType())
- new JavaRDD(newRdd)(f.elementType())
+ def fn = (a: Int, b: Iterator[T]) => f.call(a, asJavaIterator(b)).asScala
+ val newRdd = rdd.mapPartitionsWithIndex(fn, preservesPartitioning)
+ new JavaRDD(newRdd)(fakeClassTag)
}
/**

0 comments on commit 8bfd3f3

Please sign in to comment.
Something went wrong with that request. Please try again.