You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "zshao9 (via GitHub)" <gi...@apache.org> on 2024/05/22 03:50:50 UTC
[PR] [SPARK-48380][Core][WIP]: SerDeUtil.javaToPython to support batchSize parameter [spark]
zshao9 opened a new pull request, #46697:
URL: https://github.com/apache/spark/pull/46697
### What changes were proposed in this pull request?
This simple PR adds a parameter `batchSize` to the underlying `javaToPython` function. This parameter is already available on `pairRDDToPython`.
### Why are the changes needed?
With this, pyspark program can decide to use a different batch size when the default AutoBatchPickler ends up with 2GB exceeded error as in the JIRA ticket: https://issues.apache.org/jira/browse/SPARK-48380
### Does this PR introduce _any_ user-facing change?
This introduces a new way to access python RDD from python DataFrame:
```
# python
# The existing way to convert dataframe to RDD with auto batching (NO CHANGES)
df.rdd
# The new way to convert dataframe to RDD with batchSize of 1 (New API)
df._jdf.javaToPython(1)
```
### How was this patch tested?
mvn test
(still testing)
### Was this patch authored or co-authored using generative AI tooling?
No.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48380][Core][WIP]: SerDeUtil.javaToPython to support batchSize parameter [spark]
Posted by "zshao9 (via GitHub)" <gi...@apache.org>.
zshao9 commented on code in PR #46697:
URL: https://github.com/apache/spark/pull/46697#discussion_r1610389431
##########
core/src/main/scala/org/apache/spark/api/python/SerDeUtil.scala:
##########
@@ -104,12 +104,37 @@ private[spark] object SerDeUtil extends Logging {
}
}
+ /**
+ * Use a fixed batch size
+ */
+ private[spark] class FixedBatchedPickler(iter: Iterator[Any], batch: Integer) extends Iterator[Array[Byte]] {
Review Comment:
Will do!
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48380][CORE][WIP]: SerDeUtil.javaToPython to support batchSize parameter [spark]
Posted by "zshao9 (via GitHub)" <gi...@apache.org>.
zshao9 commented on code in PR #46697:
URL: https://github.com/apache/spark/pull/46697#discussion_r1612223970
##########
core/src/test/scala/org/apache/spark/api/python/SerDeUtilSuite.scala:
##########
@@ -32,5 +32,22 @@ class SerDeUtilSuite extends SparkFunSuite with SharedSparkContext {
val pythonRdd = SerDeUtil.javaToPython(javaRdd)
SerDeUtil.pythonToPairRDD(pythonRdd, false)
}
+
+ test("Converting JavaRDD to PythonRDD with a fixed batch size of 1 (SPARK-48380)") {
+ val rddWith3Elements = sc.makeRDD(Seq[(Any, Any)]((1, 2), (3, 4), (5, 6)))
+ val javaRdd = rddWith3Elements.toJavaRDD()
+ val pythonRdd = SerDeUtil.javaToPython(javaRdd, 1)
+ val collected = pythonRdd.collect()
+ assert(collected.size() === 3)
+ }
+
+ test("Converting JavaRDD to PythonRDD with an auto batch size (SPARK-48380)") {
+ val rddWith3Elements = sc.makeRDD(Seq[(Any, Any)]((1, 2), (3, 4), (5, 6)))
+ val javaRdd = rddWith3Elements.toJavaRDD()
+ val pythonRdd = SerDeUtil.javaToPython(javaRdd)
+ val collected = pythonRdd.collect()
+ assert(collected.size() === 2)
Review Comment:
This is actually failing. Need to find a way to verify that AutoBatchedPickler did batch rows together.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48380][Core][WIP]: SerDeUtil.javaToPython to support batchSize parameter [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #46697:
URL: https://github.com/apache/spark/pull/46697#discussion_r1609739643
##########
core/src/main/scala/org/apache/spark/api/python/SerDeUtil.scala:
##########
@@ -104,12 +104,37 @@ private[spark] object SerDeUtil extends Logging {
}
}
+ /**
+ * Use a fixed batch size
+ */
+ private[spark] class FixedBatchedPickler(iter: Iterator[Any], batch: Integer) extends Iterator[Array[Byte]] {
+ private val pickle = new Pickler(/* useMemo = */ true,
+ /* valueCompare = */ false)
+ private val buffer = new mutable.ArrayBuffer[Any]
+
+ override def hasNext: Boolean = iter.hasNext
+
+ override def next(): Array[Byte] = {
+ while (iter.hasNext && buffer.length < batch) {
+ buffer += iter.next()
+ }
+ val bytes = pickle.dumps(buffer.toArray)
+ buffer.clear()
+ bytes
+ }
+ }
+
/**
* Convert an RDD of Java objects to an RDD of serialized Python objects, that is usable by
* PySpark.
+ * @param batchSize if non-zero, the fixed batch size to use when pickling objects
*/
- def javaToPython(jRDD: JavaRDD[_]): JavaRDD[Array[Byte]] = {
- jRDD.rdd.mapPartitions { iter => new AutoBatchedPickler(iter) }
+ def javaToPython(jRDD: JavaRDD[_], batchSize: Integer = 0): JavaRDD[Array[Byte]] = {
+ if (batchSize == 0) {
Review Comment:
shall we fail if `batchSize < 0`?
##########
core/src/main/scala/org/apache/spark/api/python/SerDeUtil.scala:
##########
@@ -104,12 +104,37 @@ private[spark] object SerDeUtil extends Logging {
}
}
+ /**
+ * Use a fixed batch size
+ */
+ private[spark] class FixedBatchedPickler(iter: Iterator[Any], batch: Integer) extends Iterator[Array[Byte]] {
+ private val pickle = new Pickler(/* useMemo = */ true,
+ /* valueCompare = */ false)
+ private val buffer = new mutable.ArrayBuffer[Any]
+
+ override def hasNext: Boolean = iter.hasNext
+
+ override def next(): Array[Byte] = {
+ while (iter.hasNext && buffer.length < batch) {
+ buffer += iter.next()
+ }
+ val bytes = pickle.dumps(buffer.toArray)
+ buffer.clear()
+ bytes
+ }
+ }
+
/**
* Convert an RDD of Java objects to an RDD of serialized Python objects, that is usable by
* PySpark.
+ * @param batchSize if non-zero, the fixed batch size to use when pickling objects
*/
- def javaToPython(jRDD: JavaRDD[_]): JavaRDD[Array[Byte]] = {
- jRDD.rdd.mapPartitions { iter => new AutoBatchedPickler(iter) }
+ def javaToPython(jRDD: JavaRDD[_], batchSize: Integer = 0): JavaRDD[Array[Byte]] = {
Review Comment:
```suggestion
def javaToPython(jRDD: JavaRDD[_], batchSize: int = 0): JavaRDD[Array[Byte]] = {
```
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48380][CORE][WIP]: SerDeUtil.javaToPython to support rows up to 2GB [spark]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #46697: [SPARK-48380][CORE][WIP]: SerDeUtil.javaToPython to support rows up to 2GB
URL: https://github.com/apache/spark/pull/46697
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48380][CORE][WIP]: SerDeUtil.javaToPython to support batchSize parameter [spark]
Posted by "zshao9 (via GitHub)" <gi...@apache.org>.
zshao9 commented on code in PR #46697:
URL: https://github.com/apache/spark/pull/46697#discussion_r1617933166
##########
core/src/main/scala/org/apache/spark/api/python/SerDeUtil.scala:
##########
@@ -104,12 +104,40 @@ private[spark] object SerDeUtil extends Logging {
}
}
+ /**
+ * Use a fixed batch size
+ */
+ private[spark] class FixedBatchedPickler(iter: Iterator[Any], batch: Int) extends Iterator[Array[Byte]] {
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48380][Core][WIP]: SerDeUtil.javaToPython to support batchSize parameter [spark]
Posted by "zshao9 (via GitHub)" <gi...@apache.org>.
zshao9 commented on code in PR #46697:
URL: https://github.com/apache/spark/pull/46697#discussion_r1610389074
##########
core/src/main/scala/org/apache/spark/api/python/SerDeUtil.scala:
##########
@@ -104,12 +104,37 @@ private[spark] object SerDeUtil extends Logging {
}
}
+ /**
+ * Use a fixed batch size
+ */
+ private[spark] class FixedBatchedPickler(iter: Iterator[Any], batch: Integer) extends Iterator[Array[Byte]] {
+ private val pickle = new Pickler(/* useMemo = */ true,
+ /* valueCompare = */ false)
+ private val buffer = new mutable.ArrayBuffer[Any]
+
+ override def hasNext: Boolean = iter.hasNext
+
+ override def next(): Array[Byte] = {
+ while (iter.hasNext && buffer.length < batch) {
+ buffer += iter.next()
+ }
+ val bytes = pickle.dumps(buffer.toArray)
+ buffer.clear()
+ bytes
+ }
+ }
+
/**
* Convert an RDD of Java objects to an RDD of serialized Python objects, that is usable by
* PySpark.
+ * @param batchSize if non-zero, the fixed batch size to use when pickling objects
*/
- def javaToPython(jRDD: JavaRDD[_]): JavaRDD[Array[Byte]] = {
- jRDD.rdd.mapPartitions { iter => new AutoBatchedPickler(iter) }
+ def javaToPython(jRDD: JavaRDD[_], batchSize: Integer = 0): JavaRDD[Array[Byte]] = {
Review Comment:
Will do
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48380][CORE][WIP]: SerDeUtil.javaToPython to support batchSize parameter [spark]
Posted by "JoshRosen (via GitHub)" <gi...@apache.org>.
JoshRosen commented on code in PR #46697:
URL: https://github.com/apache/spark/pull/46697#discussion_r1613915291
##########
core/src/main/scala/org/apache/spark/api/python/SerDeUtil.scala:
##########
@@ -104,12 +104,40 @@ private[spark] object SerDeUtil extends Logging {
}
}
+ /**
+ * Use a fixed batch size
+ */
+ private[spark] class FixedBatchedPickler(iter: Iterator[Any], batch: Int) extends Iterator[Array[Byte]] {
Review Comment:
Looks like Scalastyle is complaining about line length, so you might need to wrap this like
```suggestion
private[spark] class FixedBatchedPickler(
iter: Iterator[Any],
batch: Int)
extends Iterator[Array[Byte]] {
```
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48380][CORE][WIP]: SerDeUtil.javaToPython to support batchSize parameter [spark]
Posted by "zshao9 (via GitHub)" <gi...@apache.org>.
zshao9 commented on code in PR #46697:
URL: https://github.com/apache/spark/pull/46697#discussion_r1619042510
##########
core/src/test/scala/org/apache/spark/api/python/SerDeUtilSuite.scala:
##########
@@ -32,5 +32,22 @@ class SerDeUtilSuite extends SparkFunSuite with SharedSparkContext {
val pythonRdd = SerDeUtil.javaToPython(javaRdd)
SerDeUtil.pythonToPairRDD(pythonRdd, false)
}
+
+ test("Converting JavaRDD to PythonRDD with a fixed batch size of 1 (SPARK-48380)") {
+ val rddWith3Elements = sc.makeRDD(Seq[(Any, Any)]((1, 2), (3, 4), (5, 6)))
+ val javaRdd = rddWith3Elements.toJavaRDD()
+ val pythonRdd = SerDeUtil.javaToPython(javaRdd, 1)
+ val collected = pythonRdd.collect()
+ assert(collected.size() === 3)
+ }
+
+ test("Converting JavaRDD to PythonRDD with an auto batch size (SPARK-48380)") {
+ val rddWith3Elements = sc.makeRDD(Seq[(Any, Any)]((1, 2), (3, 4), (5, 6)))
+ val javaRdd = rddWith3Elements.toJavaRDD()
+ val pythonRdd = SerDeUtil.javaToPython(javaRdd)
+ val collected = pythonRdd.collect()
+ assert(collected.size() === 2)
Review Comment:
Fixed by using a different test
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48380][Core][WIP]: SerDeUtil.javaToPython to support batchSize parameter [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #46697:
URL: https://github.com/apache/spark/pull/46697#discussion_r1609739061
##########
core/src/main/scala/org/apache/spark/api/python/SerDeUtil.scala:
##########
@@ -104,12 +104,37 @@ private[spark] object SerDeUtil extends Logging {
}
}
+ /**
+ * Use a fixed batch size
+ */
+ private[spark] class FixedBatchedPickler(iter: Iterator[Any], batch: Integer) extends Iterator[Array[Byte]] {
Review Comment:
```suggestion
private[spark] class FixedBatchedPickler(iter: Iterator[Any], batch: int) extends Iterator[Array[Byte]] {
```
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48380][Core][WIP]: SerDeUtil.javaToPython to support batchSize parameter [spark]
Posted by "zshao9 (via GitHub)" <gi...@apache.org>.
zshao9 commented on code in PR #46697:
URL: https://github.com/apache/spark/pull/46697#discussion_r1610397804
##########
core/src/main/scala/org/apache/spark/api/python/SerDeUtil.scala:
##########
@@ -104,12 +104,37 @@ private[spark] object SerDeUtil extends Logging {
}
}
+ /**
+ * Use a fixed batch size
+ */
+ private[spark] class FixedBatchedPickler(iter: Iterator[Any], batch: Integer) extends Iterator[Array[Byte]] {
+ private val pickle = new Pickler(/* useMemo = */ true,
+ /* valueCompare = */ false)
+ private val buffer = new mutable.ArrayBuffer[Any]
+
+ override def hasNext: Boolean = iter.hasNext
+
+ override def next(): Array[Byte] = {
+ while (iter.hasNext && buffer.length < batch) {
+ buffer += iter.next()
+ }
+ val bytes = pickle.dumps(buffer.toArray)
+ buffer.clear()
+ bytes
+ }
+ }
+
/**
* Convert an RDD of Java objects to an RDD of serialized Python objects, that is usable by
* PySpark.
+ * @param batchSize if non-zero, the fixed batch size to use when pickling objects
*/
- def javaToPython(jRDD: JavaRDD[_]): JavaRDD[Array[Byte]] = {
- jRDD.rdd.mapPartitions { iter => new AutoBatchedPickler(iter) }
+ def javaToPython(jRDD: JavaRDD[_], batchSize: Integer = 0): JavaRDD[Array[Byte]] = {
+ if (batchSize == 0) {
Review Comment:
Will do
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-48380][CORE][WIP]: SerDeUtil.javaToPython to support rows up to 2GB [spark]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #46697:
URL: https://github.com/apache/spark/pull/46697#issuecomment-2334963227
We're closing this PR because it hasn't been updated in a while. This isn't a judgement on the merit of the PR in any way. It's just a way of keeping the PR queue manageable.
If you'd like to revive this PR, please reopen it and ask a committer to remove the Stale tag!
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org