[GitHub] [spark] oleg-smith commented on a change in pull request #28488: [SPARK-29083][CORE] Prefetch elements in rdd.toLocalIterator
oleg-smith commented on a change in pull request #28488:
URL: https://github.com/apache/spark/pull/28488#discussion_r424796523
##
File path:
core/src/main/scala/org/apache/spark/rdd/util/PrefetchingIterator.scala
##
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.rdd.util
+
+import java.util.concurrent.atomic.AtomicReference
+import java.util.concurrent.locks.ReentrantLock
+
+import scala.concurrent.ExecutionContext.Implicits.global
+import scala.reflect.ClassTag
+import scala.util.Try
+
+import org.apache.spark.rdd.RDD
+
+/**
+ * Iterator that optionally prefetches next partition asynchronously
+ */
+private[spark] class PrefetchingIterator[T : ClassTag](
+rdd: RDD[T],
+prefetch: Boolean)
+ extends Iterator[Array[T]]
+with Serializable {
+
+ private val partitionIterator = rdd.partitions.indices.iterator
+
+ private val lock = new ReentrantLock()
Review comment:
since job is executed in other thread, the need of it is to
thread-safely pass the job result back to iterator
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] oleg-smith commented on a change in pull request #28488: [SPARK-29083][CORE] Prefetch elements in rdd.toLocalIterator
oleg-smith commented on a change in pull request #28488:
URL: https://github.com/apache/spark/pull/28488#discussion_r424796523
##
File path:
core/src/main/scala/org/apache/spark/rdd/util/PrefetchingIterator.scala
##
@@ -0,0 +1,103 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.rdd.util
+
+import java.util.concurrent.atomic.AtomicReference
+import java.util.concurrent.locks.ReentrantLock
+
+import scala.concurrent.ExecutionContext.Implicits.global
+import scala.reflect.ClassTag
+import scala.util.Try
+
+import org.apache.spark.rdd.RDD
+
+/**
+ * Iterator that optionally prefetches next partition asynchronously
+ */
+private[spark] class PrefetchingIterator[T : ClassTag](
+rdd: RDD[T],
+prefetch: Boolean)
+ extends Iterator[Array[T]]
+with Serializable {
+
+ private val partitionIterator = rdd.partitions.indices.iterator
+
+ private val lock = new ReentrantLock()
Review comment:
since job is executed in other thread, the need is it to thread-safely
pass the job result back to iterator
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] oleg-smith commented on a change in pull request #28488: [SPARK-29083][CORE] Prefetch elements in rdd.toLocalIterator
oleg-smith commented on a change in pull request #28488:
URL: https://github.com/apache/spark/pull/28488#discussion_r424796155
##
File path: core/src/main/scala/org/apache/spark/rdd/RDD.scala
##
@@ -1031,20 +1033,22 @@ abstract class RDD[T: ClassTag](
Array.concat(results: _*)
}
+ def toLocalIterator : Iterator[T] = toLocalIterator(false)
+
/**
* Return an iterator that contains all of the elements in this RDD.
*
* The iterator will consume as much memory as the largest partition in this
RDD.
+ * With prefetch it may consume up to the memory of the 2 largest partitions.
+ *
+ * @param prefetchPartitions If Spark should pre-fetch the next partition
before it is needed.
*
* @note This results in multiple Spark jobs, and if the input RDD is the
result
* of a wide transformation (e.g. join with different partitioners), to avoid
* recomputing the input RDD should be cached first.
*/
- def toLocalIterator: Iterator[T] = withScope {
-def collectPartition(p: Int): Array[T] = {
- sc.runJob(this, (iter: Iterator[T]) => iter.toArray, Seq(p)).head
-}
-partitions.indices.iterator.flatMap(i => collectPartition(i))
+ def toLocalIterator(prefetchPartitions: Boolean = false): Iterator[T] =
withScope {
+new PrefetchingIterator(this, prefetchPartitions).flatMap(data => data)
Review comment:
we need to flatten Iterator[array[T][ to Iterator[T]
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] oleg-smith commented on a change in pull request #28488: [SPARK-29083][CORE] Prefetch elements in rdd.toLocalIterator
oleg-smith commented on a change in pull request #28488:
URL: https://github.com/apache/spark/pull/28488#discussion_r424795937
##
File path: core/src/main/scala/org/apache/spark/rdd/RDD.scala
##
@@ -1031,20 +1033,22 @@ abstract class RDD[T: ClassTag](
Array.concat(results: _*)
}
+ def toLocalIterator : Iterator[T] = toLocalIterator(false)
+
/**
* Return an iterator that contains all of the elements in this RDD.
*
* The iterator will consume as much memory as the largest partition in this
RDD.
+ * With prefetch it may consume up to the memory of the 2 largest partitions.
+ *
+ * @param prefetchPartitions If Spark should pre-fetch the next partition
before it is needed.
*
* @note This results in multiple Spark jobs, and if the input RDD is the
result
* of a wide transformation (e.g. join with different partitioners), to avoid
* recomputing the input RDD should be cached first.
*/
- def toLocalIterator: Iterator[T] = withScope {
-def collectPartition(p: Int): Array[T] = {
- sc.runJob(this, (iter: Iterator[T]) => iter.toArray, Seq(p)).head
-}
-partitions.indices.iterator.flatMap(i => collectPartition(i))
+ def toLocalIterator(prefetchPartitions: Boolean = false): Iterator[T] =
withScope {
Review comment:
the problem is the existing method is without ().
if i reused it and added default value, it'd bring source incompatibility
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] oleg-smith commented on a change in pull request #28488: [SPARK-29083][CORE] Prefetch elements in rdd.toLocalIterator
oleg-smith commented on a change in pull request #28488:
URL: https://github.com/apache/spark/pull/28488#discussion_r424795564
##
File path: core/src/main/scala/org/apache/spark/api/java/JavaRDDLike.scala
##
@@ -368,6 +368,9 @@ trait JavaRDDLike[T, This <: JavaRDDLike[T, This]] extends
Serializable {
def toLocalIterator(): JIterator[T] =
asJavaIteratorConverter(rdd.toLocalIterator).asJava
+ def toLocalIterator(prefetchPartitions: Boolean): JIterator[T] =
Review comment:
I named if after @holdenk 's naming in Python. If you insist, I will
rename
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] oleg-smith commented on a change in pull request #28488: [SPARK-29083][CORE] Prefetch elements in rdd.toLocalIterator
oleg-smith commented on a change in pull request #28488:
URL: https://github.com/apache/spark/pull/28488#discussion_r422698223
##
File path: core/src/main/scala/org/apache/spark/rdd/RDD.scala
##
@@ -1031,20 +1033,24 @@ abstract class RDD[T: ClassTag](
Array.concat(results: _*)
}
+ def toLocalIterator : Iterator[T] = toLocalIterator(false)
+
/**
* Return an iterator that contains all of the elements in this RDD.
*
* The iterator will consume as much memory as the largest partition in this
RDD.
+ * With prefetch it may consume up to the memory of the 2 largest partitions.
+ *
+ * @param prefetchPartitions If Spark should pre-fetch the next partition
before it is needed.
*
* @note This results in multiple Spark jobs, and if the input RDD is the
result
* of a wide transformation (e.g. join with different partitioners), to avoid
* recomputing the input RDD should be cached first.
*/
- def toLocalIterator: Iterator[T] = withScope {
-def collectPartition(p: Int): Array[T] = {
- sc.runJob(this, (iter: Iterator[T]) => iter.toArray, Seq(p)).head
-}
-partitions.indices.iterator.flatMap(i => collectPartition(i))
+ def toLocalIterator(prefetchPartitions: Boolean = false): Iterator[T] =
withScope {
+val iterator = new PrefetchingIterator(this, prefetchPartitions)
+if (prefetchPartitions) iterator.hasNext
Review comment:
@holdenk @srowen Should we not prefetch the very first element here?
For a user, it might be better to get iterator without side effects first,
and submit jobs or receive errors only if iteration was intentionally started.
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] oleg-smith commented on a change in pull request #28488: [SPARK-29083][CORE] Prefetch elements in rdd.toLocalIterator
oleg-smith commented on a change in pull request #28488: URL: https://github.com/apache/spark/pull/28488#discussion_r422680143 ## File path: core/src/test/java/test/org/apache/spark/JavaAPISuite.java ## @@ -81,6 +51,26 @@ import org.apache.spark.storage.StorageLevel; import org.apache.spark.util.LongAccumulator; import org.apache.spark.util.StatCounter; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import scala.Tuple2; +import scala.Tuple3; +import scala.Tuple4; +import scala.collection.JavaConverters; + +import java.io.*; Review comment: fixed This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org - To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org For additional commands, e-mail: reviews-h...@spark.apache.org
