You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2022/10/01 00:42:40 UTC
[GitHub] [spark] liuzqt opened a new pull request, #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
liuzqt opened a new pull request, #38064:
URL: https://github.com/apache/spark/pull/38064
### What changes were proposed in this pull request?
Single task result must fit in 2GB, because we're using byte array or `ByteBuffer`(which is backed by byte array as well), and thus has a limit of 2GB(java array size limit on `byte[]`).
This PR is trying to fix this by replacing byte array with `ChunkedByteBuffer`.
### Why are the changes needed?
To overcome the 2GB limit for single task.
### Does this PR introduce _any_ user-facing change?
No
### How was this patch tested?
Unit 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
[GitHub] [spark] liuzqt commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
liuzqt commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r991524342
##########
core/src/main/scala/org/apache/spark/executor/Executor.scala:
##########
@@ -659,25 +660,27 @@ private[spark] class Executor(
val accumUpdates = task.collectAccumulatorUpdates()
val metricPeaks = metricsPoller.getTaskMetricPeaks(taskId)
// TODO: do not serialize value twice
- val directResult = new DirectTaskResult(valueBytes, accumUpdates, metricPeaks)
- val serializedDirectResult = ser.serialize(directResult)
- val resultSize = serializedDirectResult.limit()
+ val directResult = new DirectTaskResult(valueByteBuffer, accumUpdates, metricPeaks)
+ val serializedDirectResult = SerializerHelper.serializeToChunkedBuffer(ser, directResult)
+ val resultSize = serializedDirectResult.size
// directSend = sending directly back to the driver
- val serializedResult: ByteBuffer = {
+ val serializedResult: ChunkedByteBuffer = {
if (maxResultSize > 0 && resultSize > maxResultSize) {
logWarning(s"Finished $taskName. Result is larger than maxResultSize " +
s"(${Utils.bytesToString(resultSize)} > ${Utils.bytesToString(maxResultSize)}), " +
s"dropping it.")
- ser.serialize(new IndirectTaskResult[Any](TaskResultBlockId(taskId), resultSize))
+ SerializerHelper.serializeToChunkedBuffer(ser,
+ new IndirectTaskResult[Any](TaskResultBlockId(taskId), resultSize))
Review Comment:
Good point. The `serializedDirectResult` in [line 668](https://github.com/apache/spark/pull/38064/files#diff-d7a989c491f3cb77cca02c701496a9e2a3443f70af73b0d1ab0899239f3a789dR668) is the one that might exceed 2GB and we want to replace it with `ChunkedByteBuffer`.
While the `serializedResult` here is always guaranteed to be smaller than 2GB in all cases
- larger than `maxResultSize`, dropping, sending a `IndirectTaskResult` indication failure, which is small
- larger than `maxDirectResultSize`, sending a `IndirectTaskResult` with the blockId, which is small
- directly send back to driver, which is always smaller than 2GB, guarded by direct result size value check I added [here](https://github.com/apache/spark/pull/38064/files#diff-888b19a7ec330288db98c7cb699b26bcb8667bb9eba2533019f9d67b597fab95R805)
So let's just use `ByteBuffer` for `serializedResult`.
Thanks for the point!
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r994998073
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -42,8 +42,9 @@ import org.apache.spark.util.Utils
* buffers may also be used elsewhere then the caller is responsible for copying
* them as needed.
*/
-private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
+private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) extends Externalizable {
require(chunks != null, "chunks must not be null")
+ require(!chunks.contains(null), "chunks must not contain null")
Review Comment:
The next should have implicitly checked for it - was this getting triggered ?
--
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
[GitHub] [spark] Ngone51 commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r994179914
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -84,6 +92,64 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
}
}
+ /**
+ * write to stream with zero copy if possible
+ */
+ def writeToStream(out: OutputStream): Unit = {
+ var buffer: Array[Byte] = null
+ val bufferLen = 1024 * 1024
+ writeBufferToDest(this, out.write)
+ }
+
+ /**
+ * write to ObjectOutput with zero copy if possible
+ */
+ override def writeExternal(out: ObjectOutput): Unit = {
+ // we want to keep the chunks layout
+ out.writeInt(chunks.length)
+ chunks.foreach(buffer => out.writeInt(buffer.limit()))
+ chunks.foreach(buffer => out.writeBoolean(buffer.isDirect))
+ writeBufferToDest(this, out.write)
+ }
+
+ override def readExternal(in: ObjectInput): Unit = {
+ val chunksNum = in.readInt()
+ val indices = 0 until chunksNum
+ val chunksSize = indices.map(_ => in.readInt())
+ val chunksDirect = indices.map(_ => in.readBoolean())
Review Comment:
> Do we have cases where we want to preserve whether the buffer was direct or not across VM ?
I don't have such cases in my mind..
--
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
[GitHub] [spark] liuzqt commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
liuzqt commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r1004839015
##########
core/src/main/scala/org/apache/spark/executor/Executor.scala:
##########
@@ -659,9 +659,10 @@ private[spark] class Executor(
val accumUpdates = task.collectAccumulatorUpdates()
val metricPeaks = metricsPoller.getTaskMetricPeaks(taskId)
// TODO: do not serialize value twice
- val directResult = new DirectTaskResult(valueBytes, accumUpdates, metricPeaks)
- val serializedDirectResult = ser.serialize(directResult)
- val resultSize = serializedDirectResult.limit()
+ val directResult = new DirectTaskResult(valueByteBuffer, accumUpdates, metricPeaks)
+ val serializedDirectResult = SerializerHelper.serializeToChunkedBuffer(ser, directResult,
+ valueByteBuffer.size)
Review Comment:
Make senses, updated accordingly, thanks!
--
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
[GitHub] [spark] liuzqt commented on pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
liuzqt commented on PR #38064:
URL: https://github.com/apache/spark/pull/38064#issuecomment-1304306588
@mridulm I got a error when running that command in my local
```
[error] /Users/ziqi.liu/spark/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/Cast.scala:51: File line length exceeds 100 characters
```
but it seems irrelevant to this PR....and I just merged with upstream master
--
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
[GitHub] [spark] liuzqt commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
liuzqt commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r1001070001
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -84,6 +91,57 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
}
}
+ /**
+ * write to ObjectOutput with zero copy if possible
Review Comment:
Sorry for the confusion, the zero copy is actually "avoid extra copy".
- When we have access to the underlying byte array, we only need one copy: `underlying bytes array --> ObjectOutput`.
- On the other hand, when we don't have access to the underlying byte array(when it's off-heap), we need an extra copy: `bytes buffer --> temp buffer ---> ObjectOutput`.
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r1001237268
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -207,6 +267,18 @@ private[spark] object ChunkedByteBuffer {
}
out.toChunkedByteBuffer
}
+
+ /**
+ * Try to estimate appropriate chunk size so that it's not too large(waste memory) or too
+ * small(too many segments)
+ */
+ def estimateBufferChunkSize(estimatedSize: Long = -1): Int = {
+ if (estimatedSize < 0) {
+ CHUNK_BUFFER_SIZE
+ } else {
+ Math.max(Math.min(estimatedSize >> 3, CHUNK_BUFFER_SIZE).toInt, MINIMUM_CHUNK_BUFFER_SIZE)
Review Comment:
I would focus more on reasonable estimation and preventing orders of magnitude wastage.
The cost between a 8k and 10k buffer, in this context, is negligible if not zero.
--
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
[GitHub] [spark] asfgit closed pull request #38064: [SPARK-40622][SQL][CORE]Remove the limitation that single task result must fit in 2GB
Posted by GitBox <gi...@apache.org>.
asfgit closed pull request #38064: [SPARK-40622][SQL][CORE]Remove the limitation that single task result must fit in 2GB
URL: https://github.com/apache/spark/pull/38064
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r993005806
##########
core/src/main/scala/org/apache/spark/serializer/SerializerHelper.scala:
##########
@@ -0,0 +1,45 @@
+/*
+ * 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.serializer
+
+import java.nio.ByteBuffer
+
+import scala.reflect.ClassTag
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.util.io.{ChunkedByteBuffer, ChunkedByteBufferOutputStream}
+
+private[spark] object SerializerHelper extends Logging {
+ def serializeToChunkedBuffer[T: ClassTag](
+ serializerInstance: SerializerInstance,
+ t: T): ChunkedByteBuffer = {
+ val cbbos = new ChunkedByteBufferOutputStream(1024 * 1024, ByteBuffer.allocate)
Review Comment:
Review note: The default size here is fairly large - most results are fairly small (except for a small subset).
Unfortunately, small `chunkSize` leads to suboptimal performance for `ChunkedByteBufferOutputStream` - this change has a potential for causing performance regressions - something to watch out for.
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r994995056
##########
core/src/main/scala/org/apache/spark/executor/Executor.scala:
##########
@@ -659,9 +659,9 @@ private[spark] class Executor(
val accumUpdates = task.collectAccumulatorUpdates()
val metricPeaks = metricsPoller.getTaskMetricPeaks(taskId)
// TODO: do not serialize value twice
- val directResult = new DirectTaskResult(valueBytes, accumUpdates, metricPeaks)
- val serializedDirectResult = ser.serialize(directResult)
- val resultSize = serializedDirectResult.limit()
+ val directResult = new DirectTaskResult(valueByteBuffer, accumUpdates, metricPeaks)
+ val serializedDirectResult = SerializerHelper.serializeToChunkedBuffer(ser, directResult)
Review Comment:
Yes, we do know `valueByteBuffer.size` - and we know what the serialization overhead of `DirectTaskResult` is.
Will give us a much tighter bound on the size.
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r995000085
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -84,6 +91,74 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
}
}
+ /**
+ * write to ObjectOutput with zero copy if possible
+ */
+ override def writeExternal(out: ObjectOutput): Unit = {
+ // we want to keep the chunks layout
+ out.writeInt(chunks.length)
+ chunks.foreach(buffer => out.writeInt(buffer.limit()))
+ chunks.foreach(buffer => out.writeBoolean(buffer.isDirect))
+ var buffer: Array[Byte] = null
+ val bufferLen = ChunkedByteBuffer.COPY_BUFFER_LEN
+
+ getChunks().foreach { chunk => {
Review Comment:
Let us do the `getChunks` at beginning of the method - and use that for the rest of the method (instead of `this.chunks`).
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r995002920
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -172,6 +247,8 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
private[spark] object ChunkedByteBuffer {
+ val COPY_BUFFER_LEN: Int = 1024 * 1024
Review Comment:
Might be messy to add a dependency from here to `SerializerHelper` - add a comment in both places to indicate they should be ideally kept in sync ?
--
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
[GitHub] [spark] liuzqt commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
liuzqt commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r991524342
##########
core/src/main/scala/org/apache/spark/executor/Executor.scala:
##########
@@ -659,25 +660,27 @@ private[spark] class Executor(
val accumUpdates = task.collectAccumulatorUpdates()
val metricPeaks = metricsPoller.getTaskMetricPeaks(taskId)
// TODO: do not serialize value twice
- val directResult = new DirectTaskResult(valueBytes, accumUpdates, metricPeaks)
- val serializedDirectResult = ser.serialize(directResult)
- val resultSize = serializedDirectResult.limit()
+ val directResult = new DirectTaskResult(valueByteBuffer, accumUpdates, metricPeaks)
+ val serializedDirectResult = SerializerHelper.serializeToChunkedBuffer(ser, directResult)
+ val resultSize = serializedDirectResult.size
// directSend = sending directly back to the driver
- val serializedResult: ByteBuffer = {
+ val serializedResult: ChunkedByteBuffer = {
if (maxResultSize > 0 && resultSize > maxResultSize) {
logWarning(s"Finished $taskName. Result is larger than maxResultSize " +
s"(${Utils.bytesToString(resultSize)} > ${Utils.bytesToString(maxResultSize)}), " +
s"dropping it.")
- ser.serialize(new IndirectTaskResult[Any](TaskResultBlockId(taskId), resultSize))
+ SerializerHelper.serializeToChunkedBuffer(ser,
+ new IndirectTaskResult[Any](TaskResultBlockId(taskId), resultSize))
Review Comment:
Good point. The `serializedDirectResult` in [line 668](https://github.com/apache/spark/pull/38064/files#diff-d7a989c491f3cb77cca02c701496a9e2a3443f70af73b0d1ab0899239f3a789dR668) is the one that might exceed 2GB and we want to replace it with `ChunkedByteBuffer`.
While the `serializedResult` here is always guaranteed to be smaller than 2GB in all cases
- larger than `maxResultSize`, dropping, sending a `IndirectTaskResult` indication failure, which is small
- larger than `maxDirectResultSize`, sending a `IndirectTaskResult` with the blockId, which is small
- directly send back to driver, which is always small than 2GB, guarded by direct result size value check I added [here](https://github.com/apache/spark/pull/38064/files#diff-888b19a7ec330288db98c7cb699b26bcb8667bb9eba2533019f9d67b597fab95R805)
So let's just use `ByteBuffer` for `serializedResult`.
Thanks for the point!
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r994998073
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -42,8 +42,9 @@ import org.apache.spark.util.Utils
* buffers may also be used elsewhere then the caller is responsible for copying
* them as needed.
*/
-private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
+private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) extends Externalizable {
require(chunks != null, "chunks must not be null")
+ require(!chunks.contains(null), "chunks must not contain null")
Review Comment:
The next line should have implicitly checked for it - was this getting triggered ?
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r994998984
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -56,7 +57,13 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
/**
* This size of this buffer, in bytes.
*/
- val size: Long = chunks.map(_.limit().asInstanceOf[Long]).sum
+ var _size: Long = chunks.map(_.limit().asInstanceOf[Long]).sum
Review Comment:
`private` ?
--
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
[GitHub] [spark] liuzqt commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
liuzqt commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r991517511
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -84,6 +91,104 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
}
}
+ /**
+ * write to stream with zero copy if possible
+ */
+ def writeToStream(out: OutputStream): Unit = {
+ var buffer: Array[Byte] = null
+ val bufferLen = 1024 * 1024;
+ chunks.foreach(chunk => {
+ if (chunk.hasArray) {
+ // zero copy if the bytebuffer is backed by array
+ out.write(chunk.array(), chunk.arrayOffset(), chunk.limit())
+ } else {
+ // fallback to copy approach
+ if (buffer == null) {
+ buffer = new Array[Byte](bufferLen)
+ }
+ val originalPos = chunk.position()
+ chunk.rewind()
+ var bytesToRead = Math.min(chunk.remaining(), bufferLen)
+ while (bytesToRead > 0) {
+ chunk.get(buffer, 0, bytesToRead)
+ out.write(buffer, 0, bytesToRead)
+ bytesToRead = Math.min(chunk.remaining(), bufferLen)
+ }
+ chunk.position(originalPos)
Review Comment:
Good idea! Extracted to `writeBufferToDest` helper function in companion object.
--
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
[GitHub] [spark] sadikovi commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
sadikovi commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r1001233230
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -54,9 +55,16 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
private[this] var disposed: Boolean = false
/**
- * This size of this buffer, in bytes.
+ * This size of this buffer, in bytes. Using var here for serialization purpose(need to set a
Review Comment:
nit: space before paren.
##########
sql/core/src/test/scala/org/apache/spark/sql/DatasetSuite.scala:
##########
@@ -2181,6 +2184,29 @@ class DatasetSuite extends QueryTest
}
}
+class DatasetLargeResultCollectingSuite extends QueryTest
+ with SharedSparkSession {
+
+ override protected def sparkConf: SparkConf = super.sparkConf.set(MAX_RESULT_SIZE.key, "4g")
+ test("collect data with single partition larger than 2GB bytes array limit") {
+ import org.apache.spark.sql.functions.udf
+
+ val genData = udf((id: Long, bytesSize: Int) => {
+ val rand = new Random(id)
+ val arr = new Array[Byte](bytesSize)
+ rand.nextBytes(arr)
+ arr
+ })
+
+ spark.udf.register("genData", genData.asNondeterministic())
+ // create data of size ~3GB in single partition, which exceeds the byte array limit
+ // random gen to make sure it's poorly compressed
+ val df = spark.range(0, 3000, 1, 1).selectExpr("id", s"genData(id, 1000000) as data")
+ val res = df.queryExecution.executedPlan.executeCollect()
+ }
+
Review Comment:
nit: no new line.
##########
sql/core/src/test/scala/org/apache/spark/sql/DatasetSuite.scala:
##########
@@ -2181,6 +2184,29 @@ class DatasetSuite extends QueryTest
}
}
+class DatasetLargeResultCollectingSuite extends QueryTest
+ with SharedSparkSession {
+
+ override protected def sparkConf: SparkConf = super.sparkConf.set(MAX_RESULT_SIZE.key, "4g")
+ test("collect data with single partition larger than 2GB bytes array limit") {
Review Comment:
nit: new line.
##########
core/src/main/scala/org/apache/spark/serializer/SerializerHelper.scala:
##########
@@ -0,0 +1,51 @@
+/*
+ * 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.serializer
+
+import java.nio.ByteBuffer
+
+import scala.reflect.ClassTag
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.util.io.{ChunkedByteBuffer, ChunkedByteBufferOutputStream}
+
+private[spark] object SerializerHelper extends Logging {
+
+ /**
+ * @param estimatedSize estimated size of `t`, used as a hint to choose proper chunk size
Review Comment:
Can you extend the javadoc? There are 3 arguments.
##########
core/src/main/scala/org/apache/spark/serializer/SerializerHelper.scala:
##########
@@ -0,0 +1,51 @@
+/*
+ * 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.serializer
+
+import java.nio.ByteBuffer
+
+import scala.reflect.ClassTag
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.util.io.{ChunkedByteBuffer, ChunkedByteBufferOutputStream}
+
+private[spark] object SerializerHelper extends Logging {
+
+ /**
+ * @param estimatedSize estimated size of `t`, used as a hint to choose proper chunk size
+ */
+ def serializeToChunkedBuffer[T: ClassTag](
+ serializerInstance: SerializerInstance,
+ t: T,
Review Comment:
This is not an API override, so it would be good to update.
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -207,6 +267,18 @@ private[spark] object ChunkedByteBuffer {
}
out.toChunkedByteBuffer
}
+
+ /**
+ * Try to estimate appropriate chunk size so that it's not too large(waste memory) or too
+ * small(too many segments)
+ */
+ def estimateBufferChunkSize(estimatedSize: Long = -1): Int = {
+ if (estimatedSize < 0) {
+ CHUNK_BUFFER_SIZE
+ } else {
+ Math.max(Math.min(estimatedSize >> 3, CHUNK_BUFFER_SIZE).toInt, MINIMUM_CHUNK_BUFFER_SIZE)
Review Comment:
I don't think it is even worth it. You will only select a different from CHUNK_BUFFER_SIZE size when it is between CHUNK_BUFFER_SIZE and MINIMUM_CHUNK_BUFFER_SIZE and in that case you can choose either anyway.
--
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
[GitHub] [spark] liuzqt commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
liuzqt commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r1004842292
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -207,6 +267,18 @@ private[spark] object ChunkedByteBuffer {
}
out.toChunkedByteBuffer
}
+
+ /**
+ * Try to estimate appropriate chunk size so that it's not too large(waste memory) or too
+ * small(too many segments)
+ */
+ def estimateBufferChunkSize(estimatedSize: Long = -1): Int = {
+ if (estimatedSize < 0) {
+ CHUNK_BUFFER_SIZE
+ } else {
+ Math.max(Math.min(estimatedSize >> 3, CHUNK_BUFFER_SIZE).toInt, MINIMUM_CHUNK_BUFFER_SIZE)
Review Comment:
Changed to `estimatedSize`, that makes sense and should be good in this case. Thanks for the suggestions!
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r1005275497
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -84,6 +91,74 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
}
}
+ /**
+ * write to ObjectOutput with zero copy if possible
+ */
+ override def writeExternal(out: ObjectOutput): Unit = {
+ // we want to keep the chunks layout
+ out.writeInt(chunks.length)
+ chunks.foreach(buffer => out.writeInt(buffer.limit()))
+ chunks.foreach(buffer => out.writeBoolean(buffer.isDirect))
+ var buffer: Array[Byte] = null
+ val bufferLen = ChunkedByteBuffer.COPY_BUFFER_LEN
+
+ getChunks().foreach { chunk => {
+ if (chunk.hasArray) {
+ // zero copy if the bytebuffer is backed by bytes array
+ out.write(chunk.array(), chunk.arrayOffset(), chunk.limit())
+ } else {
+ // fallback to copy approach
+ if (buffer == null) {
+ buffer = new Array[Byte](bufferLen)
+ }
+ var bytesToRead = Math.min(chunk.remaining(), bufferLen)
+ while (bytesToRead > 0) {
+ chunk.get(buffer, 0, bytesToRead)
+ out.write(buffer, 0, bytesToRead)
+ bytesToRead = Math.min(chunk.remaining(), bufferLen)
+ }
Review Comment:
There is `Utils.writeByteBuffer` which writes a buffer to an `OutputStream`.
It does not do a very good job, since it allocates a buffer the same size as `remaining` - instead, we should enhance it to do what this method is doing.
Additionally, we can use a `ThreadLocal[Array[Byte]]` in `Utils` for use with this copy (the `buffer` here)
--
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
[GitHub] [spark] LuciferYang commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Remove the limitation that single task result must fit in 2GB
Posted by GitBox <gi...@apache.org>.
LuciferYang commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r1025953966
##########
sql/core/src/test/scala/org/apache/spark/sql/DatasetSuite.scala:
##########
@@ -2228,6 +2231,31 @@ class DatasetSuite extends QueryTest
}
}
+class DatasetLargeResultCollectingSuite extends QueryTest
+ with SharedSparkSession {
+
+ override protected def sparkConf: SparkConf = super.sparkConf.set(MAX_RESULT_SIZE.key, "4g")
+ test("collect data with single partition larger than 2GB bytes array limit") {
+ // This test requires large memory and leads to OOM in Github Action so we skip it. Developer
+ // should verify it in local build.
+ assume(!sys.env.contains("GITHUB_ACTIONS"))
Review Comment:
hmm... I test this suite with Java 8/11/17 on Linux/MacOS On Apple Silicon with following commands:
- Maven:
```
build/mvn clean install -DskipTests -pl sql/core -am
build/mvn clean test -pl sql/core -Dtest=none -DwildcardSuites=org.apache.spark.sql.DatasetLargeResultCollectingSuite
```
and
```
dev/change-scala-version.sh 2.13
build/mvn clean install -DskipTests -pl sql/core -am -Pscala-2.13
build/mvn clean test -pl sql/core -Pscala-2.13 -Dtest=none -DwildcardSuites=org.apache.spark.sql.DatasetLargeResultCollectingSuite
```
- SBT:
```
build/sbt clean "sql/testOnly org.apache.spark.sql.DatasetLargeResultCollectingSuite"
```
```
dev/change-scala-version.sh 2.13
build/sbt clean "sql/testOnly org.apache.spark.sql.DatasetLargeResultCollectingSuite" -Pscala-2.13
```
All test failed with `java.lang.OutOfMemoryError: Java heap space` as follows:
```
10:19:56.910 ERROR org.apache.spark.executor.Executor: Exception in task 0.0 in stage 0.0 (TID 0)
java.lang.OutOfMemoryError: Java heap space
at java.nio.HeapByteBuffer.<init>(HeapByteBuffer.java:57)
at java.nio.ByteBuffer.allocate(ByteBuffer.java:335)
at org.apache.spark.serializer.SerializerHelper$.$anonfun$serializeToChunkedBuffer$1(SerializerHelper.scala:40)
at org.apache.spark.serializer.SerializerHelper$.$anonfun$serializeToChunkedBuffer$1$adapted(SerializerHelper.scala:40)
at org.apache.spark.serializer.SerializerHelper$$$Lambda$2321/1995130077.apply(Unknown Source)
at org.apache.spark.util.io.ChunkedByteBufferOutputStream.allocateNewChunkIfNeeded(ChunkedByteBufferOutputStream.scala:87)
at org.apache.spark.util.io.ChunkedByteBufferOutputStream.write(ChunkedByteBufferOutputStream.scala:75)
at java.io.ObjectOutputStream$BlockDataOutputStream.write(ObjectOutputStream.java:1853)
at java.io.ObjectOutputStream.write(ObjectOutputStream.java:709)
at org.apache.spark.util.Utils$.$anonfun$writeByteBuffer$1(Utils.scala:271)
at org.apache.spark.util.Utils$.$anonfun$writeByteBuffer$1$adapted(Utils.scala:271)
at org.apache.spark.util.Utils$$$Lambda$2324/69671223.apply(Unknown Source)
at org.apache.spark.util.Utils$.writeByteBufferImpl(Utils.scala:249)
at org.apache.spark.util.Utils$.writeByteBuffer(Utils.scala:271)
at org.apache.spark.util.io.ChunkedByteBuffer.$anonfun$writeExternal$2(ChunkedByteBuffer.scala:103)
at org.apache.spark.util.io.ChunkedByteBuffer.$anonfun$writeExternal$2$adapted(ChunkedByteBuffer.scala:103)
at org.apache.spark.util.io.ChunkedByteBuffer$$Lambda$2323/1073743200.apply(Unknown Source)
at scala.collection.ArrayOps$.foreach$extension(ArrayOps.scala:1328)
at org.apache.spark.util.io.ChunkedByteBuffer.writeExternal(ChunkedByteBuffer.scala:103)
at java.io.ObjectOutputStream.writeExternalData(ObjectOutputStream.java:1459)
at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1430)
at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
at java.io.ObjectOutputStream.writeArray(ObjectOutputStream.java:1378)
at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1174)
at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348)
at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:46)
at org.apache.spark.serializer.SerializerHelper$.serializeToChunkedBuffer(SerializerHelper.scala:42)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:599)
```
--
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
[GitHub] [spark] Ngone51 commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r994221660
##########
core/src/test/scala/org/apache/spark/scheduler/TaskSchedulerImplSuite.scala:
##########
@@ -761,11 +761,13 @@ class TaskSchedulerImplSuite extends SparkFunSuite with LocalSparkContext
}
// End the other task of the taskset, doesn't matter whether it succeeds or fails.
val otherTask = tasks(1)
- val result = new DirectTaskResult[Int](valueSer.serialize(otherTask.taskId), Seq(), Array())
+ val result = new DirectTaskResult[Int](valueSer.serialize(otherTask.taskId), Seq(),
+ Array[Long]())
tsm.handleSuccessfulTask(otherTask.taskId, result)
} else {
tasks.foreach { task =>
- val result = new DirectTaskResult[Int](valueSer.serialize(task.taskId), Seq(), Array())
+ val result = new DirectTaskResult[Int](valueSer.serialize(task.taskId), Seq(),
+ Array[Long]())
Review Comment:
Is it a required change (adding [Long]) now? Seems like it was Long already.
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r998960187
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -207,6 +267,18 @@ private[spark] object ChunkedByteBuffer {
}
out.toChunkedByteBuffer
}
+
+ /**
+ * Try to estimate appropriate chunk size so that it's not too large(waste memory) or too
+ * small(too many segments)
+ */
+ def estimateBufferChunkSize(estimatedSize: Long = -1): Int = {
+ if (estimatedSize < 0) {
+ CHUNK_BUFFER_SIZE
+ } else {
+ Math.max(Math.min(estimatedSize >> 3, CHUNK_BUFFER_SIZE).toInt, MINIMUM_CHUNK_BUFFER_SIZE)
Review Comment:
Why are we doing `estimatedSize >> 3` ? Why not leave it as is ?
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r998972908
##########
core/src/main/scala/org/apache/spark/executor/Executor.scala:
##########
@@ -659,9 +659,10 @@ private[spark] class Executor(
val accumUpdates = task.collectAccumulatorUpdates()
val metricPeaks = metricsPoller.getTaskMetricPeaks(taskId)
// TODO: do not serialize value twice
- val directResult = new DirectTaskResult(valueBytes, accumUpdates, metricPeaks)
- val serializedDirectResult = ser.serialize(directResult)
- val resultSize = serializedDirectResult.limit()
+ val directResult = new DirectTaskResult(valueByteBuffer, accumUpdates, metricPeaks)
+ val serializedDirectResult = SerializerHelper.serializeToChunkedBuffer(ser, directResult,
+ valueByteBuffer.size)
Review Comment:
nit:
There is no cheap way to determine what the expected size would be - given mostly the bulk of it would typically be `valueBytes`, we can reasonably extrapolate and make it closer.
Something like:
`valueBytes.size + accumUpdates.size * 32 + metricPeaks.length * 8`
(A common accumulator `LongAccumulator` size is closer to 26 or so)
--
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
[GitHub] [spark] liuzqt commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
liuzqt commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r1001123093
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -56,7 +57,13 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
/**
* This size of this buffer, in bytes.
*/
- val size: Long = chunks.map(_.limit().asInstanceOf[Long]).sum
+ private var _size: Long = chunks.map(_.limit().asInstanceOf[Long]).sum
Review Comment:
Yes, I added a comment for explanation. Thanks.
--
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
[GitHub] [spark] liuzqt commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
liuzqt commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r995233234
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -84,6 +91,74 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
}
}
+ /**
+ * write to ObjectOutput with zero copy if possible
+ */
+ override def writeExternal(out: ObjectOutput): Unit = {
+ // we want to keep the chunks layout
+ out.writeInt(chunks.length)
+ chunks.foreach(buffer => out.writeInt(buffer.limit()))
+ chunks.foreach(buffer => out.writeBoolean(buffer.isDirect))
+ var buffer: Array[Byte] = null
+ val bufferLen = ChunkedByteBuffer.COPY_BUFFER_LEN
+
+ getChunks().foreach { chunk => {
+ if (chunk.hasArray) {
+ // zero copy if the bytebuffer is backed by bytes array
+ out.write(chunk.array(), chunk.arrayOffset(), chunk.limit())
+ } else {
+ // fallback to copy approach
+ if (buffer == null) {
+ buffer = new Array[Byte](bufferLen)
+ }
+ var bytesToRead = Math.min(chunk.remaining(), bufferLen)
+ while (bytesToRead > 0) {
+ chunk.get(buffer, 0, bytesToRead)
+ out.write(buffer, 0, bytesToRead)
+ bytesToRead = Math.min(chunk.remaining(), bufferLen)
+ }
Review Comment:
There should be... but I was not able to locate them....not very familiar with the code base. Feel free to let me know if you find them
--
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
[GitHub] [spark] mridulm commented on pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
mridulm commented on PR #38064:
URL: https://github.com/apache/spark/pull/38064#issuecomment-1303008250
If the failures are unrelated to your changes, try triggering the tests again.
--
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
[GitHub] [spark] liuzqt commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
liuzqt commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r1006141237
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -84,6 +91,74 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
}
}
+ /**
+ * write to ObjectOutput with zero copy if possible
+ */
+ override def writeExternal(out: ObjectOutput): Unit = {
+ // we want to keep the chunks layout
+ out.writeInt(chunks.length)
+ chunks.foreach(buffer => out.writeInt(buffer.limit()))
+ chunks.foreach(buffer => out.writeBoolean(buffer.isDirect))
+ var buffer: Array[Byte] = null
+ val bufferLen = ChunkedByteBuffer.COPY_BUFFER_LEN
+
+ getChunks().foreach { chunk => {
+ if (chunk.hasArray) {
+ // zero copy if the bytebuffer is backed by bytes array
+ out.write(chunk.array(), chunk.arrayOffset(), chunk.limit())
+ } else {
+ // fallback to copy approach
+ if (buffer == null) {
+ buffer = new Array[Byte](bufferLen)
+ }
+ var bytesToRead = Math.min(chunk.remaining(), bufferLen)
+ while (bytesToRead > 0) {
+ chunk.get(buffer, 0, bytesToRead)
+ out.write(buffer, 0, bytesToRead)
+ bytesToRead = Math.min(chunk.remaining(), bufferLen)
+ }
Review Comment:
I tried to reuse `Utils.writeByteBuffer`, and I noticed that there're two versions of `Utils.writeByteBuffer` for `OutputStream` and `DataOutput` respectively and the code is identical so added a `Utils.writeByteBufferImpl` to extract the common logic, also added ThreadLocal[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
[GitHub] [spark] liuzqt commented on pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
liuzqt commented on PR #38064:
URL: https://github.com/apache/spark/pull/38064#issuecomment-1306681050
Finally fixed it by adding `org/apache/spark/util/io`(this should be internal) to `Unidoc.ignoreUndocumentedPackages`
--
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
[GitHub] [spark] mridulm commented on pull request #38064: [SPARK-40622][SQL][CORE]Remove the limitation that single task result must fit in 2GB
Posted by GitBox <gi...@apache.org>.
mridulm commented on PR #38064:
URL: https://github.com/apache/spark/pull/38064#issuecomment-1314875312
Looks like TPCDS is still dying ...
Can we try @HyukjinKwon's suggestion @liuzqt - it will ensure we are testing this during local builds and release votes, while not causing GA to fail.
`TPCDSQuerySuite` already uses this for not executing q72
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r1008801157
##########
core/src/main/scala/org/apache/spark/util/Utils.scala:
##########
@@ -237,34 +241,42 @@ private[spark] object Utils extends Logging {
}
}
- /**
- * Primitive often used when writing [[java.nio.ByteBuffer]] to [[java.io.DataOutput]]
- */
- def writeByteBuffer(bb: ByteBuffer, out: DataOutput): Unit = {
+ private def writeByteBufferImpl(bb: ByteBuffer, writer: (Array[Byte], Int, Int) => Unit): Unit = {
if (bb.hasArray) {
- out.write(bb.array(), bb.arrayOffset() + bb.position(), bb.remaining())
+ // Avoid extra copy if the bytebuffer is backed by bytes array
+ writer(bb.array(), bb.arrayOffset() + bb.position(), bb.remaining())
} else {
+ // Fallback to copy approach
+ val buffer = {
+ // reuse the copy buffer from thread local
+ if (copyBuffer.get() == null) {
+ copyBuffer.set(new Array[Byte](COPY_BUFFER_LEN))
Review Comment:
Override `initialValue` in the `ThreadLocal` instead
--
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
[GitHub] [spark] liuzqt commented on pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
liuzqt commented on PR #38064:
URL: https://github.com/apache/spark/pull/38064#issuecomment-1302834138
@mridulm I'm looking into it, but haven't any clue so far.....it's weird that many compilation errors seems not related to this PR, and I was able to build on my local machine, not sure how to reproduce it. Would you mind taking a look? Thanks!
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r995001092
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -84,6 +91,74 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
}
}
+ /**
+ * write to ObjectOutput with zero copy if possible
+ */
+ override def writeExternal(out: ObjectOutput): Unit = {
+ // we want to keep the chunks layout
+ out.writeInt(chunks.length)
+ chunks.foreach(buffer => out.writeInt(buffer.limit()))
+ chunks.foreach(buffer => out.writeBoolean(buffer.isDirect))
+ var buffer: Array[Byte] = null
+ val bufferLen = ChunkedByteBuffer.COPY_BUFFER_LEN
+
+ getChunks().foreach { chunk => {
+ if (chunk.hasArray) {
+ // zero copy if the bytebuffer is backed by bytes array
+ out.write(chunk.array(), chunk.arrayOffset(), chunk.limit())
+ } else {
+ // fallback to copy approach
+ if (buffer == null) {
+ buffer = new Array[Byte](bufferLen)
+ }
+ var bytesToRead = Math.min(chunk.remaining(), bufferLen)
+ while (bytesToRead > 0) {
+ chunk.get(buffer, 0, bytesToRead)
+ out.write(buffer, 0, bytesToRead)
+ bytesToRead = Math.min(chunk.remaining(), bufferLen)
+ }
Review Comment:
nit: Pull this out into a separate method ?
I will need to search our codebase, but I would have expected this snippet to be already there somewhere.
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r995002920
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -172,6 +247,8 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
private[spark] object ChunkedByteBuffer {
+ val COPY_BUFFER_LEN: Int = 1024 * 1024
Review Comment:
Might be messy to add a dependency from here to `SerializerHelper` - add a comment in both places to indicate they should be ideally kept in sync ?
And make both of them private to the class ?
--
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
[GitHub] [spark] liuzqt commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
liuzqt commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r995230484
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -56,7 +57,13 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
/**
* This size of this buffer, in bytes.
*/
- val size: Long = chunks.map(_.limit().asInstanceOf[Long]).sum
+ var _size: Long = chunks.map(_.limit().asInstanceOf[Long]).sum
Review Comment:
updated
--
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
[GitHub] [spark] liuzqt commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
liuzqt commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r993750676
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -207,6 +273,41 @@ private[spark] object ChunkedByteBuffer {
}
out.toChunkedByteBuffer
}
+
+ /**
+ * util function writing ChunkedByteBuffer to destination with zero copy if possible(when
+ * ChunkedByteBuffer is backed by bytes array)
+ * @param src ChunkedByteBuffer as src
+ * @param write write function writing data to destination, following the semantic of
+ * java.io.outputStream.write(buffer Array[Byte], off Int, len Int)
+ */
+ private def writeBufferToDest(
+ src: ChunkedByteBuffer,
+ write: (Array[Byte], Int, Int) => Unit): Unit = {
+ var buffer: Array[Byte] = null
+ val bufferLen = 1024 * 1024
+
+ src.chunks.foreach { chunk => {
+ if (chunk.hasArray) {
+ // zero copy if the bytebuffer is backed by bytes array
+ write(chunk.array(), chunk.arrayOffset(), chunk.limit())
+ } else {
+ // fallback to copy approach
+ if (buffer == null) {
+ buffer = new Array[Byte](bufferLen)
+ }
+ val originalPos = chunk.position()
+ chunk.rewind()
+ var bytesToRead = Math.min(chunk.remaining(), bufferLen)
+ while (bytesToRead > 0) {
+ chunk.get(buffer, 0, bytesToRead)
+ write(buffer, 0, bytesToRead)
+ bytesToRead = Math.min(chunk.remaining(), bufferLen)
+ }
+ chunk.position(originalPos)
+ }
+ }}
Review Comment:
this method (`writeToDest`) was removed together with `writeToStream` since we don't have write-to-stream use case. Originally this method is introduced to extract common logic in writing to `ObjectOutput` and `OutputStream`, but as we don't have `OutputStream` usecase anymore, so this method can be removed. I put the implementation back to `writeExternal` instead.
--
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
[GitHub] [spark] liuzqt commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
liuzqt commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r993747322
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -84,6 +92,64 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
}
}
+ /**
+ * write to stream with zero copy if possible
+ */
+ def writeToStream(out: OutputStream): Unit = {
+ var buffer: Array[Byte] = null
+ val bufferLen = 1024 * 1024
+ writeBufferToDest(this, out.write)
+ }
Review Comment:
good catch. This method was in my first implementation and I actually don't need it any more. Removed it and also removed the `writeBufferToDest`
--
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
[GitHub] [spark] mridulm commented on pull request #38064: [SPARK-40622][SQL][CORE]Remove the limitation that single task result must fit in 2GB
Posted by GitBox <gi...@apache.org>.
mridulm commented on PR #38064:
URL: https://github.com/apache/spark/pull/38064#issuecomment-1313204753
+CC @Yikun, @HyukjinKwon - any suggestions for @liuzqt's [query above](https://github.com/apache/spark/pull/38064#issuecomment-1311348015) ? Thanks
--
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
[GitHub] [spark] liuzqt commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
liuzqt commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r993738130
##########
core/src/main/scala/org/apache/spark/serializer/SerializerHelper.scala:
##########
@@ -0,0 +1,45 @@
+/*
+ * 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.serializer
+
+import java.nio.ByteBuffer
+
+import scala.reflect.ClassTag
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.util.io.{ChunkedByteBuffer, ChunkedByteBufferOutputStream}
+
+private[spark] object SerializerHelper extends Logging {
+ def serializeToChunkedBuffer[T: ClassTag](
+ serializerInstance: SerializerInstance,
+ t: T): ChunkedByteBuffer = {
+ val cbbos = new ChunkedByteBufferOutputStream(1024 * 1024, ByteBuffer.allocate)
Review Comment:
Actually no specific reason for choosing this value, just search around `new ChunkedByteBufferOutputStream` usage, found some `1024 * 1024` in serialize manager and map output tracker, so thought it might be a reasonable value to use. Smaller size would be good for small task result, while larger chunk size is good at reducing memory fragment in case of large result but a waste for small result.
Also see my comment [below](https://github.com/apache/spark/pull/38064/#issuecomment-1276518707)(just added) about some thoughts on performance
--
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
[GitHub] [spark] liuzqt commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
liuzqt commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r995232482
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -84,6 +91,74 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
}
}
+ /**
+ * write to ObjectOutput with zero copy if possible
+ */
+ override def writeExternal(out: ObjectOutput): Unit = {
+ // we want to keep the chunks layout
+ out.writeInt(chunks.length)
+ chunks.foreach(buffer => out.writeInt(buffer.limit()))
+ chunks.foreach(buffer => out.writeBoolean(buffer.isDirect))
+ var buffer: Array[Byte] = null
+ val bufferLen = ChunkedByteBuffer.COPY_BUFFER_LEN
+
+ getChunks().foreach { chunk => {
Review Comment:
updated
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r1005277570
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -84,6 +92,64 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
}
}
+ /**
+ * write to stream with zero copy if possible
+ */
+ def writeToStream(out: OutputStream): Unit = {
+ var buffer: Array[Byte] = null
+ val bufferLen = 1024 * 1024
+ writeBufferToDest(this, out.write)
+ }
+
+ /**
+ * write to ObjectOutput with zero copy if possible
+ */
+ override def writeExternal(out: ObjectOutput): Unit = {
+ // we want to keep the chunks layout
+ out.writeInt(chunks.length)
+ chunks.foreach(buffer => out.writeInt(buffer.limit()))
+ chunks.foreach(buffer => out.writeBoolean(buffer.isDirect))
+ writeBufferToDest(this, out.write)
+ }
+
+ override def readExternal(in: ObjectInput): Unit = {
+ val chunksNum = in.readInt()
+ val indices = 0 until chunksNum
+ val chunksSize = indices.map(_ => in.readInt())
+ val chunksDirect = indices.map(_ => in.readBoolean())
Review Comment:
Looks like we are still preservin `chunksDirect`, remove it ? And always deserialize to heap buffer via `ByteBuffer.wrap` ?
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -84,6 +92,64 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
}
}
+ /**
+ * write to stream with zero copy if possible
+ */
+ def writeToStream(out: OutputStream): Unit = {
+ var buffer: Array[Byte] = null
+ val bufferLen = 1024 * 1024
+ writeBufferToDest(this, out.write)
+ }
+
+ /**
+ * write to ObjectOutput with zero copy if possible
+ */
+ override def writeExternal(out: ObjectOutput): Unit = {
+ // we want to keep the chunks layout
+ out.writeInt(chunks.length)
+ chunks.foreach(buffer => out.writeInt(buffer.limit()))
+ chunks.foreach(buffer => out.writeBoolean(buffer.isDirect))
+ writeBufferToDest(this, out.write)
+ }
+
+ override def readExternal(in: ObjectInput): Unit = {
+ val chunksNum = in.readInt()
+ val indices = 0 until chunksNum
+ val chunksSize = indices.map(_ => in.readInt())
+ val chunksDirect = indices.map(_ => in.readBoolean())
Review Comment:
Looks like we are still preserving `chunksDirect`, remove it ? And always deserialize to heap buffer via `ByteBuffer.wrap` ?
--
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
[GitHub] [spark] liuzqt commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
liuzqt commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r1001081909
##########
core/src/main/scala/org/apache/spark/serializer/SerializerHelper.scala:
##########
@@ -0,0 +1,51 @@
+/*
+ * 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.serializer
+
+import java.nio.ByteBuffer
+
+import scala.reflect.ClassTag
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.util.io.{ChunkedByteBuffer, ChunkedByteBufferOutputStream}
+
+private[spark] object SerializerHelper extends Logging {
Review Comment:
Ideally, the `SerializerInstance` should support ByteBuffer/Stream/ChunkedByteBuffer, but that would add new method to the `SerializerInstance` base class. To avoid that, I put this in a helper class (another option is to put it in `SerializerInstance`'s companion object). But I think this method should belong to serializer-related class...we can leave it here or move to `SerializerInstance`'s companion object
--
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
[GitHub] [spark] liuzqt commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
liuzqt commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r1001085443
##########
core/src/main/scala/org/apache/spark/serializer/SerializerHelper.scala:
##########
@@ -0,0 +1,51 @@
+/*
+ * 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.serializer
+
+import java.nio.ByteBuffer
+
+import scala.reflect.ClassTag
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.util.io.{ChunkedByteBuffer, ChunkedByteBufferOutputStream}
+
+private[spark] object SerializerHelper extends Logging {
+
+ /**
+ * @param estimatedSize estimated size of `t`, used as a hint to choose proper chunk size
+ */
+ def serializeToChunkedBuffer[T: ClassTag](
+ serializerInstance: SerializerInstance,
+ t: T,
Review Comment:
I was trying to follow the [signature naming in SerializerInstance](https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/serializer/Serializer.scala#L112)...
--
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
[GitHub] [spark] liuzqt commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
liuzqt commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r999794248
##########
core/src/main/scala/org/apache/spark/executor/Executor.scala:
##########
@@ -659,9 +659,10 @@ private[spark] class Executor(
val accumUpdates = task.collectAccumulatorUpdates()
val metricPeaks = metricsPoller.getTaskMetricPeaks(taskId)
// TODO: do not serialize value twice
- val directResult = new DirectTaskResult(valueBytes, accumUpdates, metricPeaks)
- val serializedDirectResult = ser.serialize(directResult)
- val resultSize = serializedDirectResult.limit()
+ val directResult = new DirectTaskResult(valueByteBuffer, accumUpdates, metricPeaks)
+ val serializedDirectResult = SerializerHelper.serializeToChunkedBuffer(ser, directResult,
+ valueByteBuffer.size)
Review Comment:
This topic is also related to the `estimatedSize >> 3` in the chunk size estimation. Since the estimated size is hard to be accurate anyway, so if we leave the chunk size to be exactly the `estimatedSize`, we're potentially wasting a large chunk (for example, the estimatedSize is 8,000, while the exact serialized size is 8,001, we're mostly wasting the last chunk of size 8000 and only contain 1 useful byte).
And dividing the chunks to size of `estimatedSize / k`, we're having a reasonably lower upper bound of the overhead(we're only wasting the last chunk of size 1000)
Given this estimation heuristic, we don't have to be super accurate(actually hardly possible to be exact) so using `valueByteBuffer.size` here for simplicity
--
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
[GitHub] [spark] mridulm commented on pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
mridulm commented on PR #38064:
URL: https://github.com/apache/spark/pull/38064#issuecomment-1311091344
We would want to make sure there are no regressions as the codebase evolves - having a unittest to capture the case would help for that.
Does local cluster with a higher memory limit help to test this ?
--
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
[GitHub] [spark] liuzqt commented on pull request #38064: [SPARK-40622][SQL][CORE]Remove the limitation that single task result must fit in 2GB
Posted by GitBox <gi...@apache.org>.
liuzqt commented on PR #38064:
URL: https://github.com/apache/spark/pull/38064#issuecomment-1314642626
Increasing jvm mem to 6G will lead to TPCDS test get killed https://github.com/liuzqt/spark/actions/runs/3464533302/jobs/5786183673
let me try 5GB...
--
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
[GitHub] [spark] LuciferYang commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Remove the limitation that single task result must fit in 2GB
Posted by GitBox <gi...@apache.org>.
LuciferYang commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r1025955547
##########
sql/core/src/test/scala/org/apache/spark/sql/DatasetSuite.scala:
##########
@@ -2228,6 +2231,31 @@ class DatasetSuite extends QueryTest
}
}
+class DatasetLargeResultCollectingSuite extends QueryTest
+ with SharedSparkSession {
+
+ override protected def sparkConf: SparkConf = super.sparkConf.set(MAX_RESULT_SIZE.key, "4g")
+ test("collect data with single partition larger than 2GB bytes array limit") {
+ // This test requires large memory and leads to OOM in Github Action so we skip it. Developer
+ // should verify it in local build.
+ assume(!sys.env.contains("GITHUB_ACTIONS"))
Review Comment:
@HyukjinKwon @mridulm @jiangxb1987 @sadikovi @Ngone51 I failed to run this case locally with Spark default Java options. Could you run successfully? Or should we ignore this case by default?
--
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
[GitHub] [spark] AmplabJenkins commented on pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on PR #38064:
URL: https://github.com/apache/spark/pull/38064#issuecomment-1264242760
Can one of the admins verify this patch?
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r998972908
##########
core/src/main/scala/org/apache/spark/executor/Executor.scala:
##########
@@ -659,9 +659,10 @@ private[spark] class Executor(
val accumUpdates = task.collectAccumulatorUpdates()
val metricPeaks = metricsPoller.getTaskMetricPeaks(taskId)
// TODO: do not serialize value twice
- val directResult = new DirectTaskResult(valueBytes, accumUpdates, metricPeaks)
- val serializedDirectResult = ser.serialize(directResult)
- val resultSize = serializedDirectResult.limit()
+ val directResult = new DirectTaskResult(valueByteBuffer, accumUpdates, metricPeaks)
+ val serializedDirectResult = SerializerHelper.serializeToChunkedBuffer(ser, directResult,
+ valueByteBuffer.size)
Review Comment:
nit:
There is no cheap way to determine what the expected size would be - given mostly the bulk of it would typically be `valueBytes`, though we can reasonably extrapolate and make it closer.
Something like:
`valueBytes.size + accumUpdates.size * 32 + metricPeaks.length * 8`
(A common accumulator `LongAccumulator` size is closer to 26 or so)
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r998974852
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -42,8 +42,9 @@ import org.apache.spark.util.Utils
* buffers may also be used elsewhere then the caller is responsible for copying
* them as needed.
*/
-private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
+private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) extends Externalizable {
require(chunks != null, "chunks must not be null")
+ require(!chunks.contains(null), "chunks must not contain null")
Review Comment:
It is another iteration over `chunks` as part of construction. Not a big deal, but I was looking to minimize impact of the change.
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r993007045
##########
core/src/main/scala/org/apache/spark/executor/Executor.scala:
##########
@@ -659,9 +659,9 @@ private[spark] class Executor(
val accumUpdates = task.collectAccumulatorUpdates()
val metricPeaks = metricsPoller.getTaskMetricPeaks(taskId)
// TODO: do not serialize value twice
- val directResult = new DirectTaskResult(valueBytes, accumUpdates, metricPeaks)
- val serializedDirectResult = ser.serialize(directResult)
- val resultSize = serializedDirectResult.limit()
+ val directResult = new DirectTaskResult(valueByteBuffer, accumUpdates, metricPeaks)
+ val serializedDirectResult = SerializerHelper.serializeToChunkedBuffer(ser, directResult)
Review Comment:
Unlike the earlier invocation of `serializeToChunkedBuffer` (L#599) , here we have a good estimate of the size - something to leverage and minimize the cost of `serializeToChunkedBuffer`
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -17,21 +17,22 @@
package org.apache.spark.util.io
-import java.io.{File, FileInputStream, InputStream}
+import java.io.{Externalizable, File, FileInputStream, InputStream, ObjectInput, ObjectOutput, OutputStream}
import java.nio.ByteBuffer
import java.nio.channels.WritableByteChannel
import com.google.common.io.ByteStreams
import com.google.common.primitives.UnsignedBytes
import org.apache.commons.io.IOUtils
-
import org.apache.spark.SparkEnv
+
import org.apache.spark.internal.config
import org.apache.spark.network.buffer.{FileSegmentManagedBuffer, ManagedBuffer}
import org.apache.spark.network.util.{ByteArrayWritableChannel, LimitedInputStream}
import org.apache.spark.storage.{EncryptedManagedBuffer, StorageUtils}
import org.apache.spark.unsafe.array.ByteArrayMethods
import org.apache.spark.util.Utils
+import org.apache.spark.util.io.ChunkedByteBuffer.writeBufferToDest
Review Comment:
nit: Replace this import with inline invocation
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -84,6 +92,64 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
}
}
+ /**
+ * write to stream with zero copy if possible
+ */
+ def writeToStream(out: OutputStream): Unit = {
+ var buffer: Array[Byte] = null
+ val bufferLen = 1024 * 1024
+ writeBufferToDest(this, out.write)
+ }
+
+ /**
+ * write to ObjectOutput with zero copy if possible
+ */
+ override def writeExternal(out: ObjectOutput): Unit = {
+ // we want to keep the chunks layout
+ out.writeInt(chunks.length)
+ chunks.foreach(buffer => out.writeInt(buffer.limit()))
+ chunks.foreach(buffer => out.writeBoolean(buffer.isDirect))
+ writeBufferToDest(this, out.write)
+ }
+
+ override def readExternal(in: ObjectInput): Unit = {
+ val chunksNum = in.readInt()
+ val indices = 0 until chunksNum
+ val chunksSize = indices.map(_ => in.readInt())
+ val chunksDirect = indices.map(_ => in.readBoolean())
Review Comment:
Do we have cases where we want to preserve whether the buffer was direct or not across VM ? The current usecase does not require it ?
+CC @Ngone51 ?
If not, drop this and simplify the impl ?
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -84,6 +92,64 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
}
}
+ /**
+ * write to stream with zero copy if possible
+ */
+ def writeToStream(out: OutputStream): Unit = {
+ var buffer: Array[Byte] = null
+ val bufferLen = 1024 * 1024
+ writeBufferToDest(this, out.write)
+ }
+
+ /**
+ * write to ObjectOutput with zero copy if possible
+ */
+ override def writeExternal(out: ObjectOutput): Unit = {
+ // we want to keep the chunks layout
+ out.writeInt(chunks.length)
+ chunks.foreach(buffer => out.writeInt(buffer.limit()))
+ chunks.foreach(buffer => out.writeBoolean(buffer.isDirect))
+ writeBufferToDest(this, out.write)
+ }
+
+ override def readExternal(in: ObjectInput): Unit = {
+ val chunksNum = in.readInt()
+ val indices = 0 until chunksNum
+ val chunksSize = indices.map(_ => in.readInt())
+ val chunksDirect = indices.map(_ => in.readBoolean())
+ val chunks = new Array[ByteBuffer](chunksNum)
+
+ val copyBufferLen = 1024 * 1024
Review Comment:
Pull this out as a constant - both `serializeToChunkedBuffer` and this method is relying on this value implicitly for efficiency.
##########
core/src/main/scala/org/apache/spark/serializer/SerializerHelper.scala:
##########
@@ -0,0 +1,45 @@
+/*
+ * 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.serializer
+
+import java.nio.ByteBuffer
+
+import scala.reflect.ClassTag
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.util.io.{ChunkedByteBuffer, ChunkedByteBufferOutputStream}
+
+private[spark] object SerializerHelper extends Logging {
+ def serializeToChunkedBuffer[T: ClassTag](
+ serializerInstance: SerializerInstance,
+ t: T): ChunkedByteBuffer = {
+ val cbbos = new ChunkedByteBufferOutputStream(1024 * 1024, ByteBuffer.allocate)
Review Comment:
The default size here is fairly large - most results are fairly small (except for a small subset).
Unfortunately, small `chunkSize` leads to suboptimal performance for `ChunkedByteBufferOutputStream` - this change has a potential for causing performance regressions - something to watch out for.
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -207,6 +273,41 @@ private[spark] object ChunkedByteBuffer {
}
out.toChunkedByteBuffer
}
+
+ /**
+ * util function writing ChunkedByteBuffer to destination with zero copy if possible(when
+ * ChunkedByteBuffer is backed by bytes array)
+ * @param src ChunkedByteBuffer as src
+ * @param write write function writing data to destination, following the semantic of
+ * java.io.outputStream.write(buffer Array[Byte], off Int, len Int)
+ */
+ private def writeBufferToDest(
+ src: ChunkedByteBuffer,
+ write: (Array[Byte], Int, Int) => Unit): Unit = {
+ var buffer: Array[Byte] = null
+ val bufferLen = 1024 * 1024
+
+ src.chunks.foreach { chunk => {
Review Comment:
Use getChunks instead of chunks directly and simplify the impl.
The `rewind` below, for example, is modifying the buffer directly.
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -207,6 +273,41 @@ private[spark] object ChunkedByteBuffer {
}
out.toChunkedByteBuffer
}
+
+ /**
+ * util function writing ChunkedByteBuffer to destination with zero copy if possible(when
+ * ChunkedByteBuffer is backed by bytes array)
+ * @param src ChunkedByteBuffer as src
+ * @param write write function writing data to destination, following the semantic of
+ * java.io.outputStream.write(buffer Array[Byte], off Int, len Int)
+ */
+ private def writeBufferToDest(
+ src: ChunkedByteBuffer,
+ write: (Array[Byte], Int, Int) => Unit): Unit = {
+ var buffer: Array[Byte] = null
+ val bufferLen = 1024 * 1024
+
+ src.chunks.foreach { chunk => {
+ if (chunk.hasArray) {
+ // zero copy if the bytebuffer is backed by bytes array
+ write(chunk.array(), chunk.arrayOffset(), chunk.limit())
+ } else {
+ // fallback to copy approach
+ if (buffer == null) {
+ buffer = new Array[Byte](bufferLen)
+ }
+ val originalPos = chunk.position()
+ chunk.rewind()
+ var bytesToRead = Math.min(chunk.remaining(), bufferLen)
+ while (bytesToRead > 0) {
+ chunk.get(buffer, 0, bytesToRead)
+ write(buffer, 0, bytesToRead)
+ bytesToRead = Math.min(chunk.remaining(), bufferLen)
+ }
+ chunk.position(originalPos)
+ }
+ }}
Review Comment:
Move this as `writeFully(out: OutputStream)` in `ChunkedByteBuffer` ?
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -84,6 +92,64 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
}
}
+ /**
+ * write to stream with zero copy if possible
+ */
+ def writeToStream(out: OutputStream): Unit = {
+ var buffer: Array[Byte] = null
+ val bufferLen = 1024 * 1024
+ writeBufferToDest(this, out.write)
+ }
Review Comment:
Why do we need this method ?
If not required, removing it will simplify `writeExternal`
--
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
[GitHub] [spark] liuzqt commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
liuzqt commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r994897386
##########
core/src/test/scala/org/apache/spark/scheduler/TaskSchedulerImplSuite.scala:
##########
@@ -761,11 +761,13 @@ class TaskSchedulerImplSuite extends SparkFunSuite with LocalSparkContext
}
// End the other task of the taskset, doesn't matter whether it succeeds or fails.
val otherTask = tasks(1)
- val result = new DirectTaskResult[Int](valueSer.serialize(otherTask.taskId), Seq(), Array())
+ val result = new DirectTaskResult[Int](valueSer.serialize(otherTask.taskId), Seq(),
+ Array[Long]())
tsm.handleSuccessfulTask(otherTask.taskId, result)
} else {
tasks.foreach { task =>
- val result = new DirectTaskResult[Int](valueSer.serialize(task.taskId), Seq(), Array())
+ val result = new DirectTaskResult[Int](valueSer.serialize(task.taskId), Seq(),
+ Array[Long]())
Review Comment:
If I remove `[Long]`, will get a compile error:
```
overloaded method constructor DirectTaskResult with alternatives:
(valueByteBuffer: java.nio.ByteBuffer,accumUpdates: Seq[org.apache.spark.util.AccumulatorV2[_, _]],metricPeaks: Array[Long])org.apache.spark.scheduler.DirectTaskResult[Int] <and>
(valueByteBuffer: org.apache.spark.util.io.ChunkedByteBuffer,accumUpdates: Seq[org.apache.spark.util.AccumulatorV2[_, _]],metricPeaks: Array[Long])org.apache.spark.scheduler.DirectTaskResult[Int]
cannot be applied to (java.nio.ByteBuffer, Seq[Nothing], Array[Nothing])
val result = new DirectTaskResult[Int](valueSer.serialize(task.taskId), Seq(),
```
The compiler somehow failed to infer the type. This happened after I added the overloaded constructor in `DirectTaskResult`.
My guessing is that overloading has to be resolved before type constraints can be determined for each parameter, those that drive type inference but overloading resolution can't happen without knowing the types of each parameter.....I haven't figured out a way to solve this
--
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
[GitHub] [spark] liuzqt commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
liuzqt commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r994995995
##########
sql/core/src/test/scala/org/apache/spark/sql/DatasetSuite.scala:
##########
@@ -2181,6 +2184,29 @@ class DatasetSuite extends QueryTest
}
}
+class DatasetLargeResultCollectingSuite extends QueryTest
+ with SharedSparkSession {
+
+ override protected def sparkConf: SparkConf = super.sparkConf.set(MAX_RESULT_SIZE.key, "4g")
+ test("collect data with single partition larger than 2GB bytes array limit") {
+ import org.apache.spark.sql.functions.udf
+
+ val genData = udf((id: Long, bytesSize: Int) => {
+ val rand = new Random(id)
+ val arr = new Array[Byte](bytesSize)
+ rand.nextBytes(arr)
+ arr
+ })
+
+ spark.udf.register("genData", genData.asNondeterministic())
+ // create data of size ~3GB in single partition, which exceeds the byte array limit
+ // random gen to make sure it's poorly compressed
+ val df = spark.range(0, 3000, 1, 1).selectExpr("id", s"genData(id, 1000000) as data")
+ val res = df.queryExecution.executedPlan.executeCollect()
Review Comment:
Yes. We want to test both serialization(executor side)/deserialization(driver side) handle the large data(>2GB limit) properly
--
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
[GitHub] [spark] mridulm commented on pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
mridulm commented on PR #38064:
URL: https://github.com/apache/spark/pull/38064#issuecomment-1278026577
@liuzqt Most task results are very small.
We will now be over-provisioning that by a few orders of magnitude when moving to `ChunkedByteBufferOutputStream` - while a vanishingly small set of cases hit the large buffer case.
This can potentially have an impact on memory utilization at executor, and if possible look at ways to mitigate - particularly, for example, when we have a good estimate of the output size.
This is not to say I have serious concerns (we do use `ChunkedByteBufferOutputStream` with precisely that size everywhere else !) - but it is not without tradeoff.
--
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
[GitHub] [spark] liuzqt commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
liuzqt commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r995120722
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -42,8 +42,9 @@ import org.apache.spark.util.Utils
* buffers may also be used elsewhere then the caller is responsible for copying
* them as needed.
*/
-private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
+private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) extends Externalizable {
require(chunks != null, "chunks must not be null")
+ require(!chunks.contains(null), "chunks must not contain null")
Review Comment:
The `ChunkeByteBuffer` has a constructor overload `def this(byteBuffer: ByteBuffer)`, which is invoked with `null` in [the DirectTaskResult empty constructor](https://github.com/apache/spark/pull/38064/files#diff-dbad0b25e806252e4c794ac2a13776bfb50064401ee0284f9f1606b6d4d4de0dL48).
It was triggered in some temp code snippet during my dev, but should not have any usage in the current code base(at least can not be found explicitly). That empty constructor might serve as default constructor for some serialization code I guess.......After all, the code in `ChunkeByteBuffer` are assuming `chunks` doesn't contains null, so I think it's no harm to do this check at the beginning.
--
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
[GitHub] [spark] liuzqt commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
liuzqt commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r995230403
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -84,6 +92,64 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
}
}
+ /**
+ * write to stream with zero copy if possible
+ */
+ def writeToStream(out: OutputStream): Unit = {
+ var buffer: Array[Byte] = null
+ val bufferLen = 1024 * 1024
+ writeBufferToDest(this, out.write)
+ }
+
+ /**
+ * write to ObjectOutput with zero copy if possible
+ */
+ override def writeExternal(out: ObjectOutput): Unit = {
+ // we want to keep the chunks layout
+ out.writeInt(chunks.length)
+ chunks.foreach(buffer => out.writeInt(buffer.limit()))
+ chunks.foreach(buffer => out.writeBoolean(buffer.isDirect))
+ writeBufferToDest(this, out.write)
+ }
+
+ override def readExternal(in: ObjectInput): Unit = {
+ val chunksNum = in.readInt()
+ val indices = 0 until chunksNum
+ val chunksSize = indices.map(_ => in.readInt())
+ val chunksDirect = indices.map(_ => in.readBoolean())
Review Comment:
Thanks for the suggestion, updated.
--
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
[GitHub] [spark] mridulm commented on pull request #38064: [SPARK-40622][SQL][CORE]Remove the limitation that single task result must fit in 2GB
Posted by GitBox <gi...@apache.org>.
mridulm commented on PR #38064:
URL: https://github.com/apache/spark/pull/38064#issuecomment-1316229029
Merged to master.
Thanks for fixing this @liuzqt !
Thanks for the reviews @Ngone51, @sadikovi, @jiangxb1987 :-)
And thanks for help with GA @HyukjinKwon and @Yikun !
--
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
[GitHub] [spark] liuzqt commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
liuzqt commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r1004834878
##########
core/src/main/scala/org/apache/spark/serializer/SerializerHelper.scala:
##########
@@ -0,0 +1,51 @@
+/*
+ * 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.serializer
+
+import java.nio.ByteBuffer
+
+import scala.reflect.ClassTag
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.util.io.{ChunkedByteBuffer, ChunkedByteBufferOutputStream}
+
+private[spark] object SerializerHelper extends Logging {
+
+ /**
+ * @param estimatedSize estimated size of `t`, used as a hint to choose proper chunk size
Review Comment:
updated
--
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
[GitHub] [spark] liuzqt commented on pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
liuzqt commented on PR #38064:
URL: https://github.com/apache/spark/pull/38064#issuecomment-1311348015
@mridulm I've tried `local-cluster[1,1,3072]` but doesn't help, I guess. Is there any way to turn up the JVM mem in github action job?
--
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
[GitHub] [spark] liuzqt commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
liuzqt commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r994902648
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -84,6 +92,64 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
}
}
+ /**
+ * write to stream with zero copy if possible
+ */
+ def writeToStream(out: OutputStream): Unit = {
+ var buffer: Array[Byte] = null
+ val bufferLen = 1024 * 1024
+ writeBufferToDest(this, out.write)
+ }
+
+ /**
+ * write to ObjectOutput with zero copy if possible
+ */
+ override def writeExternal(out: ObjectOutput): Unit = {
+ // we want to keep the chunks layout
+ out.writeInt(chunks.length)
+ chunks.foreach(buffer => out.writeInt(buffer.limit()))
+ chunks.foreach(buffer => out.writeBoolean(buffer.isDirect))
+ writeBufferToDest(this, out.write)
+ }
+
+ override def readExternal(in: ObjectInput): Unit = {
+ val chunksNum = in.readInt()
+ val indices = 0 until chunksNum
+ val chunksSize = indices.map(_ => in.readInt())
+ val chunksDirect = indices.map(_ => in.readBoolean())
Review Comment:
Shall we default to on-heap buffer in deserialization regardless what it was before serialization? @mridulm @Ngone51
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r995000085
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -84,6 +91,74 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
}
}
+ /**
+ * write to ObjectOutput with zero copy if possible
+ */
+ override def writeExternal(out: ObjectOutput): Unit = {
+ // we want to keep the chunks layout
+ out.writeInt(chunks.length)
+ chunks.foreach(buffer => out.writeInt(buffer.limit()))
+ chunks.foreach(buffer => out.writeBoolean(buffer.isDirect))
+ var buffer: Array[Byte] = null
+ val bufferLen = ChunkedByteBuffer.COPY_BUFFER_LEN
+
+ getChunks().foreach { chunk => {
Review Comment:
Let us do the `getChunks` at beginning of the method - and use that for the rest of the method (instead of `this.chunks` above).
--
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
[GitHub] [spark] mridulm commented on pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
mridulm commented on PR #38064:
URL: https://github.com/apache/spark/pull/38064#issuecomment-1304281819
Looks like doc build is failing and so failing build ...
Can you run `build/sbt -Phadoop-3 -Pyarn -Pdocker-integration-tests -Pspark-ganglia-lgpl -Phive -Pmesos -Phive-thriftserver -Pkubernetes -Pkinesis-asl -Phadoop-cloud unidoc` against your local repo and see if it succeeds ?
Fix might be a simple case of updating to latest master
(Works fine on local build for me)
--
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
[GitHub] [spark] Yikun commented on pull request #38064: [SPARK-40622][SQL][CORE]Remove the limitation that single task result must fit in 2GB
Posted by GitBox <gi...@apache.org>.
Yikun commented on PR #38064:
URL: https://github.com/apache/spark/pull/38064#issuecomment-1313530277
The `2GB` should not the limited of github action, there only are [a 7GB limited](https://docs.github.com/en/actions/using-github-hosted-runners/about-github-hosted-runners#supported-runners-and-hardware-resources) for total memory.
I'm not sure adding some jvm mem opts by setting [sbt opt](https://github.com/apache/spark/blob/0594cad8c28c1026c838323e43a7a2ffe72350aa/build/sbt#L69-L70) could help or not.
--
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
[GitHub] [spark] Ngone51 commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r991002266
##########
core/src/main/scala/org/apache/spark/serializer/SerializerHelper.scala:
##########
@@ -0,0 +1,44 @@
+/*
+ * 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.serializer
+
+import java.nio.ByteBuffer
+
+import scala.reflect.ClassTag
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.util.io.{ChunkedByteBuffer, ChunkedByteBufferOutputStream}
+
+private[spark] object SerializerHelper extends Logging {
+ def serializeToChunkedBuffer[T: ClassTag](
+ serializerInstance: SerializerInstance,
+ t: T): ChunkedByteBuffer = {
+ val cbbos = new ChunkedByteBufferOutputStream(1024 * 1024, ByteBuffer.allocate)
+ val out = serializerInstance.serializeStream(cbbos)
+ out.writeObject(t)
+ out.close()
+ cbbos.close()
+ cbbos.toChunkedByteBuffer
+ }
Review Comment:
leave a blank line after this?
##########
core/src/main/scala/org/apache/spark/executor/Executor.scala:
##########
@@ -659,25 +660,27 @@ private[spark] class Executor(
val accumUpdates = task.collectAccumulatorUpdates()
val metricPeaks = metricsPoller.getTaskMetricPeaks(taskId)
// TODO: do not serialize value twice
- val directResult = new DirectTaskResult(valueBytes, accumUpdates, metricPeaks)
- val serializedDirectResult = ser.serialize(directResult)
- val resultSize = serializedDirectResult.limit()
+ val directResult = new DirectTaskResult(valueByteBuffer, accumUpdates, metricPeaks)
+ val serializedDirectResult = SerializerHelper.serializeToChunkedBuffer(ser, directResult)
+ val resultSize = serializedDirectResult.size
// directSend = sending directly back to the driver
- val serializedResult: ByteBuffer = {
+ val serializedResult: ChunkedByteBuffer = {
if (maxResultSize > 0 && resultSize > maxResultSize) {
logWarning(s"Finished $taskName. Result is larger than maxResultSize " +
s"(${Utils.bytesToString(resultSize)} > ${Utils.bytesToString(maxResultSize)}), " +
s"dropping it.")
- ser.serialize(new IndirectTaskResult[Any](TaskResultBlockId(taskId), resultSize))
+ SerializerHelper.serializeToChunkedBuffer(ser,
+ new IndirectTaskResult[Any](TaskResultBlockId(taskId), resultSize))
Review Comment:
The object `IndirectTaskResult` should be small. Do we also want to use chunked buffer for it?
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -42,8 +42,9 @@ import org.apache.spark.util.Utils
* buffers may also be used elsewhere then the caller is responsible for copying
* them as needed.
*/
-private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
+private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) extends Externalizable {
require(chunks != null, "chunks must not be null")
+ require(!chunks.contains(null), "chunks must not be null")
Review Comment:
```suggestion
require(!chunks.contains(null), "chunks must not contain null")
```
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -84,6 +91,104 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
}
}
+ /**
+ * write to stream with zero copy if possible
+ */
+ def writeToStream(out: OutputStream): Unit = {
+ var buffer: Array[Byte] = null
+ val bufferLen = 1024 * 1024;
+ chunks.foreach(chunk => {
+ if (chunk.hasArray) {
+ // zero copy if the bytebuffer is backed by array
+ out.write(chunk.array(), chunk.arrayOffset(), chunk.limit())
+ } else {
+ // fallback to copy approach
+ if (buffer == null) {
+ buffer = new Array[Byte](bufferLen)
+ }
+ val originalPos = chunk.position()
+ chunk.rewind()
+ var bytesToRead = Math.min(chunk.remaining(), bufferLen)
+ while (bytesToRead > 0) {
+ chunk.get(buffer, 0, bytesToRead)
+ out.write(buffer, 0, bytesToRead)
+ bytesToRead = Math.min(chunk.remaining(), bufferLen)
+ }
+ chunk.position(originalPos)
Review Comment:
Shall we extract this common code block to a function?
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -84,6 +91,104 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
}
}
+ /**
+ * write to stream with zero copy if possible
+ */
+ def writeToStream(out: OutputStream): Unit = {
+ var buffer: Array[Byte] = null
+ val bufferLen = 1024 * 1024;
+ chunks.foreach(chunk => {
+ if (chunk.hasArray) {
+ // zero copy if the bytebuffer is backed by array
+ out.write(chunk.array(), chunk.arrayOffset(), chunk.limit())
+ } else {
+ // fallback to copy approach
+ if (buffer == null) {
+ buffer = new Array[Byte](bufferLen)
+ }
+ val originalPos = chunk.position()
+ chunk.rewind()
+ var bytesToRead = Math.min(chunk.remaining(), bufferLen)
+ while (bytesToRead > 0) {
+ chunk.get(buffer, 0, bytesToRead)
+ out.write(buffer, 0, bytesToRead)
+ bytesToRead = Math.min(chunk.remaining(), bufferLen)
+ }
+ chunk.position(originalPos)
+ }
+ })
+ }
+
+ /**
+ * write to ObjectOutput with zero copy if possible
+ */
+ override def writeExternal(out: ObjectOutput): Unit = {
+ // we want to keep the chunks layout
+ out.writeInt(chunks.length)
+ chunks.foreach(buffer => out.writeInt(buffer.limit()))
+ chunks.foreach(buffer => out.writeBoolean(buffer.isDirect))
+ var buffer: Array[Byte] = null
+ val bufferLen = 1024 * 1024;
+ chunks.foreach(chunk => {
+ if (chunk.hasArray) {
+ // zero copy if the bytebuffer is backed by array
+ out.write(chunk.array(), chunk.arrayOffset(), chunk.limit())
+ } else {
+ // fallback to copy approach
+ if (buffer == null) {
+ buffer = new Array[Byte](bufferLen)
+ }
+ val originalPos = chunk.position()
+ chunk.rewind()
+ var bytesToRead = Math.min(chunk.remaining(), bufferLen)
+ while (bytesToRead > 0) {
+ chunk.get(buffer, 0, bytesToRead)
+ out.write(buffer, 0, bytesToRead)
+ bytesToRead = Math.min(chunk.remaining(), bufferLen)
+ }
+ chunk.position(originalPos)
+ }
+ })
+ }
+
+ override def readExternal(in: ObjectInput): Unit = {
+ val chunksNum = in.readInt()
+ val indices = 0 until chunksNum
+ val chunksSize = indices.map(_ => in.readInt())
+ val chunksDirect = indices.map(_ => in.readBoolean())
+ val chunks = new Array[ByteBuffer](chunksNum)
+
+ val copyBufferLen = 1024 * 1024
+ val copyBuffer: Array[Byte] = if (chunksDirect.exists(identity)) {
+ new Array[Byte](copyBufferLen)
+ } else {
+ null
+ }
+
+ indices.foreach(i => {
Review Comment:
```scala
indices.foreach { i =>
...
}
```
##########
core/src/main/scala/org/apache/spark/serializer/SerializerHelper.scala:
##########
@@ -0,0 +1,44 @@
+/*
+ * 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.serializer
+
+import java.nio.ByteBuffer
+
+import scala.reflect.ClassTag
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.util.io.{ChunkedByteBuffer, ChunkedByteBufferOutputStream}
+
+private[spark] object SerializerHelper extends Logging {
+ def serializeToChunkedBuffer[T: ClassTag](
+ serializerInstance: SerializerInstance,
+ t: T): ChunkedByteBuffer = {
+ val cbbos = new ChunkedByteBufferOutputStream(1024 * 1024, ByteBuffer.allocate)
+ val out = serializerInstance.serializeStream(cbbos)
+ out.writeObject(t)
+ out.close()
+ cbbos.close()
+ cbbos.toChunkedByteBuffer
+ }
+ def deserializeFromChunkedBuffer[T: ClassTag](
+ serializerInstance: SerializerInstance,
+ bytes: ChunkedByteBuffer): T = {
Review Comment:
4 indents
##########
core/src/main/scala/org/apache/spark/executor/Executor.scala:
##########
@@ -659,25 +660,27 @@ private[spark] class Executor(
val accumUpdates = task.collectAccumulatorUpdates()
val metricPeaks = metricsPoller.getTaskMetricPeaks(taskId)
// TODO: do not serialize value twice
- val directResult = new DirectTaskResult(valueBytes, accumUpdates, metricPeaks)
- val serializedDirectResult = ser.serialize(directResult)
- val resultSize = serializedDirectResult.limit()
+ val directResult = new DirectTaskResult(valueByteBuffer, accumUpdates, metricPeaks)
+ val serializedDirectResult = SerializerHelper.serializeToChunkedBuffer(ser, directResult)
+ val resultSize = serializedDirectResult.size
// directSend = sending directly back to the driver
- val serializedResult: ByteBuffer = {
+ val serializedResult: ChunkedByteBuffer = {
if (maxResultSize > 0 && resultSize > maxResultSize) {
logWarning(s"Finished $taskName. Result is larger than maxResultSize " +
s"(${Utils.bytesToString(resultSize)} > ${Utils.bytesToString(maxResultSize)}), " +
s"dropping it.")
- ser.serialize(new IndirectTaskResult[Any](TaskResultBlockId(taskId), resultSize))
+ SerializerHelper.serializeToChunkedBuffer(ser,
+ new IndirectTaskResult[Any](TaskResultBlockId(taskId), resultSize))
} else if (resultSize > maxDirectResultSize) {
val blockId = TaskResultBlockId(taskId)
env.blockManager.putBytes(
blockId,
- new ChunkedByteBuffer(serializedDirectResult.duplicate()),
+ serializedDirectResult,
StorageLevel.MEMORY_AND_DISK_SER)
logInfo(s"Finished $taskName. $resultSize bytes result sent via BlockManager)")
- ser.serialize(new IndirectTaskResult[Any](blockId, resultSize))
+ SerializerHelper.serializeToChunkedBuffer(ser,
+ new IndirectTaskResult[Any](blockId, resultSize))
Review Comment:
same here.
##########
core/src/main/scala/org/apache/spark/scheduler/TaskResult.scala:
##########
@@ -24,44 +24,49 @@ import scala.collection.mutable.ArrayBuffer
import org.apache.spark.SparkEnv
import org.apache.spark.metrics.ExecutorMetricType
-import org.apache.spark.serializer.SerializerInstance
+import org.apache.spark.serializer.{SerializerHelper, SerializerInstance}
import org.apache.spark.storage.BlockId
import org.apache.spark.util.{AccumulatorV2, Utils}
+import org.apache.spark.util.io.ChunkedByteBuffer
// Task result. Also contains updates to accumulator variables and executor metric peaks.
private[spark] sealed trait TaskResult[T]
/** A reference to a DirectTaskResult that has been stored in the worker's BlockManager. */
-private[spark] case class IndirectTaskResult[T](blockId: BlockId, size: Int)
+private[spark] case class IndirectTaskResult[T](blockId: BlockId, size: Long)
extends TaskResult[T] with Serializable
/** A TaskResult that contains the task's return value, accumulator updates and metric peaks. */
private[spark] class DirectTaskResult[T](
- var valueBytes: ByteBuffer,
+ var valueByteBuffer: ChunkedByteBuffer,
Review Comment:
pls keep the indentation unchanged.
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -84,6 +91,104 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
}
}
+ /**
+ * write to stream with zero copy if possible
+ */
+ def writeToStream(out: OutputStream): Unit = {
+ var buffer: Array[Byte] = null
+ val bufferLen = 1024 * 1024;
+ chunks.foreach(chunk => {
Review Comment:
nit:
```scala
chunks.foreach { chunk =>
...
}
```
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -84,6 +91,104 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
}
}
+ /**
+ * write to stream with zero copy if possible
+ */
+ def writeToStream(out: OutputStream): Unit = {
+ var buffer: Array[Byte] = null
+ val bufferLen = 1024 * 1024;
+ chunks.foreach(chunk => {
+ if (chunk.hasArray) {
+ // zero copy if the bytebuffer is backed by array
+ out.write(chunk.array(), chunk.arrayOffset(), chunk.limit())
+ } else {
+ // fallback to copy approach
+ if (buffer == null) {
+ buffer = new Array[Byte](bufferLen)
+ }
+ val originalPos = chunk.position()
+ chunk.rewind()
+ var bytesToRead = Math.min(chunk.remaining(), bufferLen)
+ while (bytesToRead > 0) {
+ chunk.get(buffer, 0, bytesToRead)
+ out.write(buffer, 0, bytesToRead)
+ bytesToRead = Math.min(chunk.remaining(), bufferLen)
+ }
+ chunk.position(originalPos)
+ }
+ })
+ }
+
+ /**
+ * write to ObjectOutput with zero copy if possible
+ */
+ override def writeExternal(out: ObjectOutput): Unit = {
+ // we want to keep the chunks layout
+ out.writeInt(chunks.length)
+ chunks.foreach(buffer => out.writeInt(buffer.limit()))
+ chunks.foreach(buffer => out.writeBoolean(buffer.isDirect))
+ var buffer: Array[Byte] = null
+ val bufferLen = 1024 * 1024;
+ chunks.foreach(chunk => {
Review Comment:
nit:
```
chunks.foreach { chunk =>
...
}
```
--
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
[GitHub] [spark] liuzqt commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
liuzqt commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r993747322
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -84,6 +92,64 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
}
}
+ /**
+ * write to stream with zero copy if possible
+ */
+ def writeToStream(out: OutputStream): Unit = {
+ var buffer: Array[Byte] = null
+ val bufferLen = 1024 * 1024
+ writeBufferToDest(this, out.write)
+ }
Review Comment:
good catch. This method is in my first implementation and I actually don't need it any more. Removed it and also remove the `writeBufferToDest`
--
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
[GitHub] [spark] LuciferYang commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Remove the limitation that single task result must fit in 2GB
Posted by GitBox <gi...@apache.org>.
LuciferYang commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r1025960041
##########
sql/core/src/test/scala/org/apache/spark/sql/DatasetSuite.scala:
##########
@@ -2228,6 +2231,31 @@ class DatasetSuite extends QueryTest
}
}
+class DatasetLargeResultCollectingSuite extends QueryTest
+ with SharedSparkSession {
+
+ override protected def sparkConf: SparkConf = super.sparkConf.set(MAX_RESULT_SIZE.key, "4g")
+ test("collect data with single partition larger than 2GB bytes array limit") {
+ // This test requires large memory and leads to OOM in Github Action so we skip it. Developer
+ // should verify it in local build.
+ assume(!sys.env.contains("GITHUB_ACTIONS"))
Review Comment:
OK
--
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
[GitHub] [spark] LuciferYang commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Remove the limitation that single task result must fit in 2GB
Posted by GitBox <gi...@apache.org>.
LuciferYang commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r1025953966
##########
sql/core/src/test/scala/org/apache/spark/sql/DatasetSuite.scala:
##########
@@ -2228,6 +2231,31 @@ class DatasetSuite extends QueryTest
}
}
+class DatasetLargeResultCollectingSuite extends QueryTest
+ with SharedSparkSession {
+
+ override protected def sparkConf: SparkConf = super.sparkConf.set(MAX_RESULT_SIZE.key, "4g")
+ test("collect data with single partition larger than 2GB bytes array limit") {
+ // This test requires large memory and leads to OOM in Github Action so we skip it. Developer
+ // should verify it in local build.
+ assume(!sys.env.contains("GITHUB_ACTIONS"))
Review Comment:
hmm... I test this suite with **Java 8/11/17** on Linux/MacOS On Apple Silicon with following commands:
- Maven:
```
build/mvn clean install -DskipTests -pl sql/core -am
build/mvn clean test -pl sql/core -Dtest=none -DwildcardSuites=org.apache.spark.sql.DatasetLargeResultCollectingSuite
```
and
```
dev/change-scala-version.sh 2.13
build/mvn clean install -DskipTests -pl sql/core -am -Pscala-2.13
build/mvn clean test -pl sql/core -Pscala-2.13 -Dtest=none -DwildcardSuites=org.apache.spark.sql.DatasetLargeResultCollectingSuite
```
- SBT:
```
build/sbt clean "sql/testOnly org.apache.spark.sql.DatasetLargeResultCollectingSuite"
```
```
dev/change-scala-version.sh 2.13
build/sbt clean "sql/testOnly org.apache.spark.sql.DatasetLargeResultCollectingSuite" -Pscala-2.13
```
All test failed with `java.lang.OutOfMemoryError: Java heap space` as follows:
```
10:19:56.910 ERROR org.apache.spark.executor.Executor: Exception in task 0.0 in stage 0.0 (TID 0)
java.lang.OutOfMemoryError: Java heap space
at java.nio.HeapByteBuffer.<init>(HeapByteBuffer.java:57)
at java.nio.ByteBuffer.allocate(ByteBuffer.java:335)
at org.apache.spark.serializer.SerializerHelper$.$anonfun$serializeToChunkedBuffer$1(SerializerHelper.scala:40)
at org.apache.spark.serializer.SerializerHelper$.$anonfun$serializeToChunkedBuffer$1$adapted(SerializerHelper.scala:40)
at org.apache.spark.serializer.SerializerHelper$$$Lambda$2321/1995130077.apply(Unknown Source)
at org.apache.spark.util.io.ChunkedByteBufferOutputStream.allocateNewChunkIfNeeded(ChunkedByteBufferOutputStream.scala:87)
at org.apache.spark.util.io.ChunkedByteBufferOutputStream.write(ChunkedByteBufferOutputStream.scala:75)
at java.io.ObjectOutputStream$BlockDataOutputStream.write(ObjectOutputStream.java:1853)
at java.io.ObjectOutputStream.write(ObjectOutputStream.java:709)
at org.apache.spark.util.Utils$.$anonfun$writeByteBuffer$1(Utils.scala:271)
at org.apache.spark.util.Utils$.$anonfun$writeByteBuffer$1$adapted(Utils.scala:271)
at org.apache.spark.util.Utils$$$Lambda$2324/69671223.apply(Unknown Source)
at org.apache.spark.util.Utils$.writeByteBufferImpl(Utils.scala:249)
at org.apache.spark.util.Utils$.writeByteBuffer(Utils.scala:271)
at org.apache.spark.util.io.ChunkedByteBuffer.$anonfun$writeExternal$2(ChunkedByteBuffer.scala:103)
at org.apache.spark.util.io.ChunkedByteBuffer.$anonfun$writeExternal$2$adapted(ChunkedByteBuffer.scala:103)
at org.apache.spark.util.io.ChunkedByteBuffer$$Lambda$2323/1073743200.apply(Unknown Source)
at scala.collection.ArrayOps$.foreach$extension(ArrayOps.scala:1328)
at org.apache.spark.util.io.ChunkedByteBuffer.writeExternal(ChunkedByteBuffer.scala:103)
at java.io.ObjectOutputStream.writeExternalData(ObjectOutputStream.java:1459)
at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1430)
at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1548)
at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1509)
at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1432)
at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1178)
at java.io.ObjectOutputStream.writeArray(ObjectOutputStream.java:1378)
at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1174)
at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:348)
at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:46)
at org.apache.spark.serializer.SerializerHelper$.serializeToChunkedBuffer(SerializerHelper.scala:42)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:599)
```
--
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
[GitHub] [spark] HyukjinKwon commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Remove the limitation that single task result must fit in 2GB
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r1026007189
##########
sql/core/src/test/scala/org/apache/spark/sql/DatasetSuite.scala:
##########
@@ -2228,6 +2231,31 @@ class DatasetSuite extends QueryTest
}
}
+class DatasetLargeResultCollectingSuite extends QueryTest
+ with SharedSparkSession {
+
+ override protected def sparkConf: SparkConf = super.sparkConf.set(MAX_RESULT_SIZE.key, "4g")
+ test("collect data with single partition larger than 2GB bytes array limit") {
+ // This test requires large memory and leads to OOM in Github Action so we skip it. Developer
+ // should verify it in local build.
+ assume(!sys.env.contains("GITHUB_ACTIONS"))
Review Comment:
Thanks man
--
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
[GitHub] [spark] HyukjinKwon commented on pull request #38064: [SPARK-40622][SQL][CORE]Remove the limitation that single task result must fit in 2GB
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on PR #38064:
URL: https://github.com/apache/spark/pull/38064#issuecomment-1314714912
I actually don't think we can have a test with 2GB because GA is already using a lot of memory in fact. One way would be to add a test, and mark it as `ignore` so other people can verify that manually in the future.
Another way is to skip the tests in GitHub Actions enviornment (e.g., `assume(!sys.env.contains("GITHUB_ACTIONS"))`)
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r1001241694
##########
core/src/main/scala/org/apache/spark/executor/Executor.scala:
##########
@@ -659,9 +659,10 @@ private[spark] class Executor(
val accumUpdates = task.collectAccumulatorUpdates()
val metricPeaks = metricsPoller.getTaskMetricPeaks(taskId)
// TODO: do not serialize value twice
- val directResult = new DirectTaskResult(valueBytes, accumUpdates, metricPeaks)
- val serializedDirectResult = ser.serialize(directResult)
- val resultSize = serializedDirectResult.limit()
+ val directResult = new DirectTaskResult(valueByteBuffer, accumUpdates, metricPeaks)
+ val serializedDirectResult = SerializerHelper.serializeToChunkedBuffer(ser, directResult,
+ valueByteBuffer.size)
Review Comment:
As I mentioned above, get reasonable upper bound - we can always improve in future.
The main reason why I suggested improving for accumulators particularly is because typically rest of DirectTaskResult in much smaller than result value - so a reasonable overestimation there might help minimize need for two large buffers when one might do.
But this is mostly a nit - we can add a comment to the code, if we are unsure of the heurestics, to improve it for future.
The main reason for suggesting this is to reduce wastage at driver.
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r998975346
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -172,6 +247,8 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
private[spark] object ChunkedByteBuffer {
+ val COPY_BUFFER_LEN: Int = 1024 * 1024
Review Comment:
Added a comment which might help ...
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r994996066
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -84,6 +92,64 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
}
}
+ /**
+ * write to stream with zero copy if possible
+ */
+ def writeToStream(out: OutputStream): Unit = {
+ var buffer: Array[Byte] = null
+ val bufferLen = 1024 * 1024
+ writeBufferToDest(this, out.write)
+ }
+
+ /**
+ * write to ObjectOutput with zero copy if possible
+ */
+ override def writeExternal(out: ObjectOutput): Unit = {
+ // we want to keep the chunks layout
+ out.writeInt(chunks.length)
+ chunks.foreach(buffer => out.writeInt(buffer.limit()))
+ chunks.foreach(buffer => out.writeBoolean(buffer.isDirect))
+ writeBufferToDest(this, out.write)
+ }
+
+ override def readExternal(in: ObjectInput): Unit = {
+ val chunksNum = in.readInt()
+ val indices = 0 until chunksNum
+ val chunksSize = indices.map(_ => in.readInt())
+ val chunksDirect = indices.map(_ => in.readBoolean())
Review Comment:
Yes, let us do that - if some specific reason comes up in future, we can add support as required.
Will make the implementation much more simpler.
--
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
[GitHub] [spark] sadikovi commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
sadikovi commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r1001117156
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -42,8 +42,9 @@ import org.apache.spark.util.Utils
* buffers may also be used elsewhere then the caller is responsible for copying
* them as needed.
*/
-private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
+private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) extends Externalizable {
Review Comment:
I understand that, I was just asking if the references are maintained outside of this class, seems like it is not an issue.
--
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
[GitHub] [spark] liuzqt commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
liuzqt commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r993746185
##########
core/src/main/scala/org/apache/spark/executor/Executor.scala:
##########
@@ -659,9 +659,9 @@ private[spark] class Executor(
val accumUpdates = task.collectAccumulatorUpdates()
val metricPeaks = metricsPoller.getTaskMetricPeaks(taskId)
// TODO: do not serialize value twice
- val directResult = new DirectTaskResult(valueBytes, accumUpdates, metricPeaks)
- val serializedDirectResult = ser.serialize(directResult)
- val resultSize = serializedDirectResult.limit()
+ val directResult = new DirectTaskResult(valueByteBuffer, accumUpdates, metricPeaks)
+ val serializedDirectResult = SerializerHelper.serializeToChunkedBuffer(ser, directResult)
Review Comment:
do you mean we can explicitly choose the chunk size here to avoid too-small/to-large chunk?
--
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
[GitHub] [spark] liuzqt commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
liuzqt commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r993757787
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -207,6 +273,41 @@ private[spark] object ChunkedByteBuffer {
}
out.toChunkedByteBuffer
}
+
+ /**
+ * util function writing ChunkedByteBuffer to destination with zero copy if possible(when
+ * ChunkedByteBuffer is backed by bytes array)
+ * @param src ChunkedByteBuffer as src
+ * @param write write function writing data to destination, following the semantic of
+ * java.io.outputStream.write(buffer Array[Byte], off Int, len Int)
+ */
+ private def writeBufferToDest(
+ src: ChunkedByteBuffer,
+ write: (Array[Byte], Int, Int) => Unit): Unit = {
+ var buffer: Array[Byte] = null
+ val bufferLen = 1024 * 1024
+
+ src.chunks.foreach { chunk => {
Review Comment:
good point, thanks! This saves me from restoring the states
--
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
[GitHub] [spark] Ngone51 commented on pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on PR #38064:
URL: https://github.com/apache/spark/pull/38064#issuecomment-1311315439
Should the PR title be changed to something like "Remove the limitation of a single task result must fit in 2GB"?
--
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
[GitHub] [spark] mridulm commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
mridulm commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r998975346
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -172,6 +247,8 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
private[spark] object ChunkedByteBuffer {
+ val COPY_BUFFER_LEN: Int = 1024 * 1024
Review Comment:
Add a comment which might help ...
--
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
[GitHub] [spark] mridulm commented on pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
mridulm commented on PR #38064:
URL: https://github.com/apache/spark/pull/38064#issuecomment-1299638357
Can you pls take a look at the build failure @liuzqt ?
--
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
[GitHub] [spark] mridulm commented on pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
mridulm commented on PR #38064:
URL: https://github.com/apache/spark/pull/38064#issuecomment-1308159592
The test failures look like due to the memory requirements for the test 'collect data with single partition larger than 2GB bytes array limit' is too high - causing OOM.
--
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
[GitHub] [spark] liuzqt commented on pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
liuzqt commented on PR #38064:
URL: https://github.com/apache/spark/pull/38064#issuecomment-1306322963
I fixed the double quote issue in `Cast` and command `build/sbt -Phadoop-3 -Pyarn -Pdocker-integration-tests -Pspark-ganglia-lgpl -Phive -Pmesos -Phive-thriftserver -Pkubernetes -Pkinesis-asl -Phadoop-cloud unidoc` works well in my local. But the build is still failing...
--
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
[GitHub] [spark] liuzqt commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
liuzqt commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r1016232819
##########
core/src/test/scala/org/apache/spark/serializer/KryoSerializerResizableOutputSuite.scala:
##########
@@ -34,18 +32,20 @@ class KryoSerializerResizableOutputSuite extends SparkFunSuite {
conf.set(SERIALIZER, "org.apache.spark.serializer.KryoSerializer")
conf.set(KRYO_SERIALIZER_BUFFER_SIZE.key, "1m")
conf.set(KRYO_SERIALIZER_MAX_BUFFER_SIZE.key, "1m")
- withSpark(new SparkContext("local", "test", conf)) { sc =>
Review Comment:
After code change, this test(`sc.parallelize(x).collect()`) doesn't go through `Serializer.serialize` code path anymore, it goes through `Serializer.serializeStream` instead. So I directly construct a serializer to test against that logic.
The reason why `Serializer.serializeStream` is not affected by buffer size is that the way `KryoSerializeStream` constructs `KryoOutput` is different from how `KyroSerializeInstance` does it:
- https://github.com/apache/spark/blob/2298cebcf83f7cad21fa291a718505032830474a/core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala#L264
- https://github.com/apache/spark/blob/2298cebcf83f7cad21fa291a718505032830474a/core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala#L380
I'm not sure this is by intention or not, but to minimize the change I leave it as it's. The test suite should also preserve the original testing semantics.
--
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
[GitHub] [spark] liuzqt commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
liuzqt commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r995232173
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -172,6 +247,8 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
private[spark] object ChunkedByteBuffer {
+ val COPY_BUFFER_LEN: Int = 1024 * 1024
Review Comment:
I added a `def estimateBufferChunkSize(estimatedSize: Long = -1)` to be used for both. But I'm not sure if the heuristic is appropriate.....
Or another option: we can use `1024`(1KB) for all, make it simple. I did some quick benchmark, 1KB isn't too bad compared to 1MB even in large result, and the overhead upper bound is reasonable even when result is very tiny(actually even a nearly empty result will still be serialized to a few hundred bytes because of other metrics and accumulators)
WDYT?
--
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
[GitHub] [spark] liuzqt commented on pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
liuzqt commented on PR #38064:
URL: https://github.com/apache/spark/pull/38064#issuecomment-1276518707
Some general comments about the performance implication regarding replacing `Array[Byte]` and `ByteBuffer`(backed by `Array[Byte]`) with `ChunkedByteBuffer`:
- when reading from stream (i.e., `ByteArrayInputStream` vs `ChunkedByteBufferInputStream`), no much differences, while `ByteArrayInputStream` might a little bit win in terms of cache locality because of continuous memory, but `ChunkedByteBuffer` won't be too bad along as the chunk is reasonable
- when we're writing to stream(i.e., `ByteArrayOutputStream` vs `ChunkedByteBufferOutputStream`)
- `ByteArrayOutputStream` start with a small buffer(32 bytes) and grow 2x exponentially, and have to do **array copy** every grow
- `ChunkedByteBufferOutputStream` use fixed `chunk size` to grow(which you can specify when you create the stream), while the grow is **append style** instead of **copy style**
- do some manual benchmark on large data, `ChunkedByteBufferOutputStream` is much faster, (tried different data size from 100MB to 1GB and different chunk size from 1KB to 1MB, can see at least ~2x speedup), I would attribute to array copy overhead mostly.
- when eventually dump to `ByteBuffer`(or raw byte array) vs. `ChunkedByteBuffer`, the latter might waste some memory space in the last chunk, but not a big deal I believe. And in serialization they're the same.
- after all, result collection is a small portion in the whole end-to-end query
--
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
[GitHub] [spark] LuciferYang commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Remove the limitation that single task result must fit in 2GB
Posted by GitBox <gi...@apache.org>.
LuciferYang commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r1025954405
##########
sql/core/src/test/scala/org/apache/spark/sql/DatasetSuite.scala:
##########
@@ -2228,6 +2231,31 @@ class DatasetSuite extends QueryTest
}
}
+class DatasetLargeResultCollectingSuite extends QueryTest
+ with SharedSparkSession {
+
+ override protected def sparkConf: SparkConf = super.sparkConf.set(MAX_RESULT_SIZE.key, "4g")
+ test("collect data with single partition larger than 2GB bytes array limit") {
+ // This test requires large memory and leads to OOM in Github Action so we skip it. Developer
+ // should verify it in local build.
+ assume(!sys.env.contains("GITHUB_ACTIONS"))
Review Comment:
@liuzqt Could you tell me how to run the local test successfully with Spark default Java options?
--
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
[GitHub] [spark] LuciferYang commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Remove the limitation that single task result must fit in 2GB
Posted by GitBox <gi...@apache.org>.
LuciferYang commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r1025982465
##########
sql/core/src/test/scala/org/apache/spark/sql/DatasetSuite.scala:
##########
@@ -2228,6 +2231,31 @@ class DatasetSuite extends QueryTest
}
}
+class DatasetLargeResultCollectingSuite extends QueryTest
+ with SharedSparkSession {
+
+ override protected def sparkConf: SparkConf = super.sparkConf.set(MAX_RESULT_SIZE.key, "4g")
+ test("collect data with single partition larger than 2GB bytes array limit") {
+ // This test requires large memory and leads to OOM in Github Action so we skip it. Developer
+ // should verify it in local build.
+ assume(!sys.env.contains("GITHUB_ACTIONS"))
Review Comment:
https://github.com/apache/spark/pull/38704 ignore this case as default
--
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
[GitHub] [spark] HyukjinKwon commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Remove the limitation that single task result must fit in 2GB
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r1025959335
##########
sql/core/src/test/scala/org/apache/spark/sql/DatasetSuite.scala:
##########
@@ -2228,6 +2231,31 @@ class DatasetSuite extends QueryTest
}
}
+class DatasetLargeResultCollectingSuite extends QueryTest
+ with SharedSparkSession {
+
+ override protected def sparkConf: SparkConf = super.sparkConf.set(MAX_RESULT_SIZE.key, "4g")
+ test("collect data with single partition larger than 2GB bytes array limit") {
+ // This test requires large memory and leads to OOM in Github Action so we skip it. Developer
+ // should verify it in local build.
+ assume(!sys.env.contains("GITHUB_ACTIONS"))
Review Comment:
Ah .. this is tricky.. Let's just probably mark this test as `ignore` for now then.. with adding some comments like some custom memory configurations would have to be applied ...
--
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
[GitHub] [spark] sadikovi commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
sadikovi commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r1000114683
##########
core/src/main/scala/org/apache/spark/serializer/SerializerHelper.scala:
##########
@@ -0,0 +1,51 @@
+/*
+ * 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.serializer
+
+import java.nio.ByteBuffer
+
+import scala.reflect.ClassTag
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.util.io.{ChunkedByteBuffer, ChunkedByteBufferOutputStream}
+
+private[spark] object SerializerHelper extends Logging {
+
+ /**
+ * @param estimatedSize estimated size of `t`, used as a hint to choose proper chunk size
+ */
+ def serializeToChunkedBuffer[T: ClassTag](
+ serializerInstance: SerializerInstance,
+ t: T,
Review Comment:
nit: could you have a bit more descriptive name? Thanks.
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -42,8 +42,9 @@ import org.apache.spark.util.Utils
* buffers may also be used elsewhere then the caller is responsible for copying
* them as needed.
*/
-private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
+private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) extends Externalizable {
Review Comment:
Can you confirm that these are copies of the byte buffers so there is no other reference that can be modified outside of the ChunkedByteBuffer based on your changes?
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -84,6 +91,57 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
}
}
+ /**
+ * write to ObjectOutput with zero copy if possible
Review Comment:
nit: `Writes to the provided ObjectOutput with zero copy if possible.`
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -207,6 +267,18 @@ private[spark] object ChunkedByteBuffer {
}
out.toChunkedByteBuffer
}
+
+ /**
+ * Try to estimate appropriate chunk size so that it's not too large(waste memory) or too
+ * small(too many segments)
Review Comment:
nit: space before (.
##########
core/src/test/scala/org/apache/spark/io/ChunkedByteBufferSuite.scala:
##########
@@ -69,6 +78,45 @@ class ChunkedByteBufferSuite extends SparkFunSuite with SharedSparkContext {
}
}
+ test("Externalizable: writeExternal() and readExternal()") {
+ // intentionally generate arrays of different len, in order to verify the chunks layout
+ // is preserved after ser/deser
+ val byteArrays = (1 to 15).map(i => (0 until i).map(_.toByte).toArray)
+ val chunkedByteBuffer = new ChunkedByteBuffer(byteArrays.map(ByteBuffer.wrap).toArray)
+ val baos = new ByteArrayOutputStream()
+ val objOut = new ObjectOutputStream(baos)
+ chunkedByteBuffer.writeExternal(objOut)
+ objOut.close()
+ assert(chunkedByteBuffer.chunks.forall(_.position() == 0))
+
+ val chunkedByteBuffer2 = {
+ val tmp = new ChunkedByteBuffer
+ tmp.readExternal(new ObjectInputStream(new ByteArrayInputStream(baos.toByteArray)))
+ tmp
+ }
+ assertBufferEqual(chunkedByteBuffer, chunkedByteBuffer2)
+ }
+
+ test(
+ "Externalizable: writeExternal() and readExternal() should handle off-heap buffer properly") {
Review Comment:
nit:
```
....
) {
val ...
```
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -84,6 +91,57 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
}
}
+ /**
+ * write to ObjectOutput with zero copy if possible
Review Comment:
Also, could you elaborate on zero copy? It seems a copy is required regardless.
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -84,6 +91,57 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
}
}
+ /**
+ * write to ObjectOutput with zero copy if possible
+ */
+ override def writeExternal(out: ObjectOutput): Unit = {
+ // we want to keep the chunks layout
+ out.writeInt(chunks.length)
+ val chunksCopy = getChunks()
+ chunksCopy.foreach(buffer => out.writeInt(buffer.limit()))
+ var buffer: Array[Byte] = null
+ val bufferLen = ChunkedByteBuffer.estimateBufferChunkSize(size)
+
+ chunksCopy.foreach { chunk => {
+ if (chunk.hasArray) {
+ // zero copy if the bytebuffer is backed by bytes array
+ out.write(chunk.array(), chunk.arrayOffset(), chunk.limit())
Review Comment:
I believe this would still incur a copy, unless the implementation stores the reference and position.
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -56,7 +57,13 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
/**
* This size of this buffer, in bytes.
*/
- val size: Long = chunks.map(_.limit().asInstanceOf[Long]).sum
+ private var _size: Long = chunks.map(_.limit().asInstanceOf[Long]).sum
Review Comment:
Is this needed for serialisation? Can you add a comment to explain why this is `var`? Thanks.
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -84,6 +91,57 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
}
}
+ /**
+ * write to ObjectOutput with zero copy if possible
+ */
+ override def writeExternal(out: ObjectOutput): Unit = {
+ // we want to keep the chunks layout
Review Comment:
nit: We.... Can you uppercase other comments in the method as well?
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -84,6 +91,57 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
}
}
+ /**
+ * write to ObjectOutput with zero copy if possible
+ */
+ override def writeExternal(out: ObjectOutput): Unit = {
+ // we want to keep the chunks layout
+ out.writeInt(chunks.length)
+ val chunksCopy = getChunks()
+ chunksCopy.foreach(buffer => out.writeInt(buffer.limit()))
+ var buffer: Array[Byte] = null
+ val bufferLen = ChunkedByteBuffer.estimateBufferChunkSize(size)
+
+ chunksCopy.foreach { chunk => {
+ if (chunk.hasArray) {
+ // zero copy if the bytebuffer is backed by bytes array
+ out.write(chunk.array(), chunk.arrayOffset(), chunk.limit())
+ } else {
+ // fallback to copy approach
+ if (buffer == null) {
+ buffer = new Array[Byte](bufferLen)
+ }
+ var bytesToRead = Math.min(chunk.remaining(), bufferLen)
+ while (bytesToRead > 0) {
+ chunk.get(buffer, 0, bytesToRead)
+ out.write(buffer, 0, bytesToRead)
+ bytesToRead = Math.min(chunk.remaining(), bufferLen)
+ }
+ }
+ }}
+ }
+
+ override def readExternal(in: ObjectInput): Unit = {
+ val chunksNum = in.readInt()
+ val indices = 0 until chunksNum
+ val chunksSize = indices.map(_ => in.readInt())
+ val chunks = new Array[ByteBuffer](chunksNum)
+
+ // We deserialize all chunks into on-heap buffer by default. If we have use case in the future
+ // where we want to preserve the on-heap/off-heap nature of chunks, then we need to record the
+ // `isDirect` property of each chunk during serialization
+ indices.foreach { i => {
Review Comment:
nit: you don't need the second `{`.
##########
core/src/test/scala/org/apache/spark/io/ChunkedByteBufferSuite.scala:
##########
@@ -28,6 +29,14 @@ import org.apache.spark.util.io.ChunkedByteBuffer
class ChunkedByteBufferSuite extends SparkFunSuite with SharedSparkContext {
+ def assertBufferEqual(buffer1: ChunkedByteBuffer, buffer2: ChunkedByteBuffer): Unit = {
+ assert(buffer1.chunks.length == buffer2.chunks.length)
+ assert(buffer1.chunks.zip(buffer2.chunks).forall{
Review Comment:
nit: space before {.
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -207,6 +267,18 @@ private[spark] object ChunkedByteBuffer {
}
out.toChunkedByteBuffer
}
+
+ /**
+ * Try to estimate appropriate chunk size so that it's not too large(waste memory) or too
Review Comment:
nit: space before (.
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -207,6 +267,18 @@ private[spark] object ChunkedByteBuffer {
}
out.toChunkedByteBuffer
}
+
+ /**
+ * Try to estimate appropriate chunk size so that it's not too large(waste memory) or too
+ * small(too many segments)
+ */
+ def estimateBufferChunkSize(estimatedSize: Long = -1): Int = {
+ if (estimatedSize < 0) {
+ CHUNK_BUFFER_SIZE
+ } else {
+ Math.max(Math.min(estimatedSize >> 3, CHUNK_BUFFER_SIZE).toInt, MINIMUM_CHUNK_BUFFER_SIZE)
Review Comment:
+1
##########
core/src/main/scala/org/apache/spark/serializer/SerializerHelper.scala:
##########
@@ -0,0 +1,51 @@
+/*
+ * 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.serializer
+
+import java.nio.ByteBuffer
+
+import scala.reflect.ClassTag
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.util.io.{ChunkedByteBuffer, ChunkedByteBufferOutputStream}
+
+private[spark] object SerializerHelper extends Logging {
Review Comment:
I was thinking if this is specific to the chunked byte buffer, would it be beneficial to add this to ChunkedByteBuffer as a static method, e.g. `ChunkedByteBuffer.serialize(...)`? Just a question to see if there is a utility in the separate class.
--
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
[GitHub] [spark] liuzqt commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
liuzqt commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r1001073725
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -84,6 +91,57 @@ private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
}
}
+ /**
+ * write to ObjectOutput with zero copy if possible
+ */
+ override def writeExternal(out: ObjectOutput): Unit = {
+ // we want to keep the chunks layout
+ out.writeInt(chunks.length)
+ val chunksCopy = getChunks()
+ chunksCopy.foreach(buffer => out.writeInt(buffer.limit()))
+ var buffer: Array[Byte] = null
+ val bufferLen = ChunkedByteBuffer.estimateBufferChunkSize(size)
+
+ chunksCopy.foreach { chunk => {
+ if (chunk.hasArray) {
+ // zero copy if the bytebuffer is backed by bytes array
+ out.write(chunk.array(), chunk.arrayOffset(), chunk.limit())
Review Comment:
I think the `array()` will returns the reference of the byte array that backs this buffer, which should not incur extra copy
--
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
[GitHub] [spark] sadikovi commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
sadikovi commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r1001117156
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -42,8 +42,9 @@ import org.apache.spark.util.Utils
* buffers may also be used elsewhere then the caller is responsible for copying
* them as needed.
*/
-private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
+private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) extends Externalizable {
Review Comment:
I understand that, I was asking if the references are maintained outside of this class.
--
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
[GitHub] [spark] liuzqt commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
liuzqt commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r1001155698
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -42,8 +42,9 @@ import org.apache.spark.util.Utils
* buffers may also be used elsewhere then the caller is responsible for copying
* them as needed.
*/
-private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
+private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) extends Externalizable {
Review Comment:
Ahh got it. Yes in the case of this PR this should not be an issue since the chunk buffer is used within a pretty short life cycle during serialization/IO and reference ownership is clear. Thanks for the question.
--
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
[GitHub] [spark] sadikovi commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
sadikovi commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r1001232748
##########
core/src/main/scala/org/apache/spark/serializer/SerializerHelper.scala:
##########
@@ -0,0 +1,51 @@
+/*
+ * 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.serializer
+
+import java.nio.ByteBuffer
+
+import scala.reflect.ClassTag
+
+import org.apache.spark.internal.Logging
+import org.apache.spark.util.io.{ChunkedByteBuffer, ChunkedByteBufferOutputStream}
+
+private[spark] object SerializerHelper extends Logging {
+
+ /**
+ * @param estimatedSize estimated size of `t`, used as a hint to choose proper chunk size
+ */
+ def serializeToChunkedBuffer[T: ClassTag](
+ serializerInstance: SerializerInstance,
+ t: T,
Review Comment:
Yeah, that is probably not a good choice there either. It is fine, let's keep as is.
--
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
[GitHub] [spark] liuzqt commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
liuzqt commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r1001055617
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -42,8 +42,9 @@ import org.apache.spark.util.Utils
* buffers may also be used elsewhere then the caller is responsible for copying
* them as needed.
*/
-private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) {
+private[spark] class ChunkedByteBuffer(var chunks: Array[ByteBuffer]) extends Externalizable {
Review Comment:
There is no physical bytes buffer copy happening here, so the `chunks` underlying bytes buffer are actually still modifiable outside, and I think it's the caller's responsibility to manage the ownership. (But it is what it was, not introduced by this PR, I think the idea of `ChunkedByteBuffer` is more like a "view" of buffer chunks)
But this PR can guarantee that the serialization won't modify any state of the ChunkedByteBuffer.
--
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
[GitHub] [spark] liuzqt commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
liuzqt commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r1001086967
##########
core/src/main/scala/org/apache/spark/util/io/ChunkedByteBuffer.scala:
##########
@@ -207,6 +267,18 @@ private[spark] object ChunkedByteBuffer {
}
out.toChunkedByteBuffer
}
+
+ /**
+ * Try to estimate appropriate chunk size so that it's not too large(waste memory) or too
+ * small(too many segments)
+ */
+ def estimateBufferChunkSize(estimatedSize: Long = -1): Int = {
+ if (estimatedSize < 0) {
+ CHUNK_BUFFER_SIZE
+ } else {
+ Math.max(Math.min(estimatedSize >> 3, CHUNK_BUFFER_SIZE).toInt, MINIMUM_CHUNK_BUFFER_SIZE)
Review Comment:
The logic is explained in the [below comments](https://github.com/apache/spark/pull/38064#discussion_r999794248), the heuristic is somehow arbitrary but I think it should make some sense. Feel free to leave any feedback, thanks!
--
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
[GitHub] [spark] mridulm commented on pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
mridulm commented on PR #38064:
URL: https://github.com/apache/spark/pull/38064#issuecomment-1305283892
The issue seems to be the use of `“` (0x201c) and `”` (0x201d) in the comment - not `"` (0x22 - double quotes).
This results in line number difference between bytes and characters.
Since this is blocking your PR, and given it is only a change in comment in `Cast`, do you mind replacing all occurances of these two characters in that file @liuzqt ?
+CC @beliefer who made the change and @cloud-fan who had merged it.
--
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
[GitHub] [spark] Ngone51 commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r994222831
##########
sql/core/src/test/scala/org/apache/spark/sql/DatasetSuite.scala:
##########
@@ -2181,6 +2184,29 @@ class DatasetSuite extends QueryTest
}
}
+class DatasetLargeResultCollectingSuite extends QueryTest
+ with SharedSparkSession {
+
+ override protected def sparkConf: SparkConf = super.sparkConf.set(MAX_RESULT_SIZE.key, "4g")
+ test("collect data with single partition larger than 2GB bytes array limit") {
+ import org.apache.spark.sql.functions.udf
+
+ val genData = udf((id: Long, bytesSize: Int) => {
+ val rand = new Random(id)
+ val arr = new Array[Byte](bytesSize)
+ rand.nextBytes(arr)
+ arr
+ })
+
+ spark.udf.register("genData", genData.asNondeterministic())
+ // create data of size ~3GB in single partition, which exceeds the byte array limit
+ // random gen to make sure it's poorly compressed
+ val df = spark.range(0, 3000, 1, 1).selectExpr("id", s"genData(id, 1000000) as data")
+ val res = df.queryExecution.executedPlan.executeCollect()
Review Comment:
Will this take much memory on the driver?
--
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
[GitHub] [spark] liuzqt commented on a diff in pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
liuzqt commented on code in PR #38064:
URL: https://github.com/apache/spark/pull/38064#discussion_r1016232819
##########
core/src/test/scala/org/apache/spark/serializer/KryoSerializerResizableOutputSuite.scala:
##########
@@ -34,18 +32,20 @@ class KryoSerializerResizableOutputSuite extends SparkFunSuite {
conf.set(SERIALIZER, "org.apache.spark.serializer.KryoSerializer")
conf.set(KRYO_SERIALIZER_BUFFER_SIZE.key, "1m")
conf.set(KRYO_SERIALIZER_MAX_BUFFER_SIZE.key, "1m")
- withSpark(new SparkContext("local", "test", conf)) { sc =>
Review Comment:
After code change, this test(`sc.parallelize(x).collect()`) doesn't go through `Serializer.serialize` code path anymore, it goes through `Serializer.serializeStream` instead. So I directly construct a serializer to test against that logic.
The reason why `Serializer.serializeStream` is not affected by buffer size is that the way `KryoSerializeStream` constructs `KryoOutput` is different from how `KyroSerializeInstance` does it:
- https://github.com/apache/spark/blob/2298cebcf83f7cad21fa291a718505032830474a/core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala#L264
- https://github.com/apache/spark/blob/2298cebcf83f7cad21fa291a718505032830474a/core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala#L380
--
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
[GitHub] [spark] liuzqt commented on pull request #38064: [SPARK-40622][SQL][CORE]Result of a single task in collect() must fit in 2GB
Posted by GitBox <gi...@apache.org>.
liuzqt commented on PR #38064:
URL: https://github.com/apache/spark/pull/38064#issuecomment-1309558197
Tried reduce the large result test to ~2.1GB, as well as local-cluster approach but neither works....Since I've verified that test on my local, so I removed the 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