[SPARK-14277] UnsafeSorterSpillReader should do buffered read from un… #12074
Conversation
…derlying compression stream
sitalkedia
force-pushed
the
bufferedReader
branch
from
b8720c4 to
5ad27f4
Compare
|
Is this the only place where buffering helps or would it make sense to do buffered reads from Snappy streams in other circumstances as well? In other words, should this buffering perhaps either be done at more call-sites of |
|
Also, /cc @xerial, who may be able to comment on whether |
|
@JoshRosen - There might be other places where buffering might help, I did not notice any other hotspot during my job run though. Also, as you mentioned pushing this into |
|
Jenkins, this is ok to test. |
|
A reason snappy-java's SnappyInputStream uses Snappy.arrayCopy (JNI method) is to load the uncompressed data into primitive type arrays (e.g., float[], int[]) since there is no standard Java method for doing this. When writing data to byte[], replacing the implementation with non-JNI based one (using System.arrayCopy) would be possible. |
|
Thanks @xerial , this is going to fix all snappy read/write inefficiency due to small writes. |
|
I have just deployed snappy-java-1.1.2.3 with this fix, which will be synchronized to the Maven central soon. |
|
That's great. Thanks a lot for the quick fix. |
|
Test build #54564 has finished for PR 12074 at commit
|
|
@JoshRosen - I guess after @xerial 's change, we won't be needing this change, right? |
|
@sitalkedia, if you confirm that the updated |
|
@JoshRosen - thanks, working on it. Will update soon. |
|
@xerial - I am seeing similar issue for snappy write as well. Can we fix the write code path as well? Stack trace - org.xerial.snappy.SnappyNative.arrayCopy(Native Method) |
|
@sitalkedia Sure. I'll do that. |
Use System.arraycopy for write(byte[]) apache/spark#12074
|
Released snappy-java-1.1.2.4 with this fix. Thanks for letting me know. |
|
Thanks @xerial! @sitalkedia, feel free to open a new PR for the dep. bump after you finish testing this new version. |
|
Great! @sitalkedia, do you mind closing this PR in favor of #12096 and updating the SPARK-14277 JIRA's description to match your new PR so that it accurately describes the change that we're going to commit? Thanks! |
|
Changed the SPARK-14277 JIRA's description, closing this PR. |
4 participants
What changes were proposed in this pull request?
While running a Spark job which is spilling a lot of data in reduce phase, we see that significant amount of CPU is being consumed in native Snappy ArrayCopy method (Please see the stack trace below).
Stack trace -
org.xerial.snappy.SnappyNative.$$YJP$$arrayCopy(Native Method)
org.xerial.snappy.SnappyNative.arrayCopy(SnappyNative.java)
org.xerial.snappy.Snappy.arrayCopy(Snappy.java:85)
org.xerial.snappy.SnappyInputStream.rawRead(SnappyInputStream.java:190)
org.xerial.snappy.SnappyInputStream.read(SnappyInputStream.java:163)
java.io.DataInputStream.readFully(DataInputStream.java:195)
java.io.DataInputStream.readLong(DataInputStream.java:416)
org.apache.spark.util.collection.unsafe.sort.UnsafeSorterSpillReader.loadNext(UnsafeSorterSpillReader.java:71)
org.apache.spark.util.collection.unsafe.sort.UnsafeSorterSpillMerger$2.loadNext(UnsafeSorterSpillMerger.java:79)
org.apache.spark.sql.execution.UnsafeExternalRowSorter$1.next(UnsafeExternalRowSorter.java:136)
org.apache.spark.sql.execution.UnsafeExternalRowSorter$1.next(UnsafeExternalRowSorter.java:123)
The reason for that is the SpillReader does a lot of small reads from the underlying snappy compressed stream and we pay a heavy cost of jni calls for these small reads. The SpillReader should instead do a buffered read from the underlying snappy compressed stream.
How was this patch tested?
Tested by running the job and we saw more than 10% cpu savings.
(If this patch involves UI changes, please attach a screenshot; otherwise, remove this)
…derlying compression stream