[SPARK-40912][CORE]Overhead of Exceptions in KryoDeserializationStream #38428

eejbyfeldt · 2022-10-28T13:02:16Z

What changes were proposed in this pull request?

This PR avoid exceptions in the implementation of KryoDeserializationStream.

Why are the changes needed?

Using an exceptions for end of stream is slow, especially for small streams. It also problematic as it the exception caught in the KryoDeserializationStream could also be caused by corrupt data which would just be ignored in the current implementation.

Does this PR introduce any user-facing change?

Yes, it changes so some method on KryoDeserializationStream no longer raises EOFException.

How was this patch tested?

Existing tests.

This PR only changes KryoDeserializationStream as a proof of concept. If this is the direction we want to go we should probably change DerserializationStream isntead so that the interface is consistent.

AmplabJenkins · 2022-10-28T14:02:32Z

Can one of the admins verify this patch?

mridulm

The change itself looks promising, thanks for working on it @eejbyfeldt !
Given DeserializationStream is a public api, I would want to be more conservative in changes to it. Let us see how to proposal develops.

core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala

mridulm · 2022-10-30T06:24:21Z

core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala

+  final override def asKeyValueIterator: Iterator[(Any, Any)] = new NextIterator[(Any, Any)] {
+    override protected def getNext() = {
+      if (KryoDeserializationStream.this.hasNext) {
+        (readKey[Any](), readValue[Any]())


Given we are fix this, not make assumptions that if key is present, value will be as well ?

You mean that if only a key exist we just ignore it like the current implementation would?

Or potentially do something better.

if (hasNext) { val key = readKey() if (hasNext) { return (key, readValue()) } }

But given this is corner case enough, I would consider this change mostly a nit.

mridulm · 2022-10-30T06:25:35Z

core/src/main/scala/org/apache/spark/serializer/KryoSerializer.scala

-      case e: KryoException
-        if e.getMessage.toLowerCase(Locale.ROOT).contains("buffer underflow") =>
-        throw new EOFException
-    }


Preserve this even with the proposed change of checking eof - to continue catching cases where EOF is encountered prematurely ?
This will be mainly to handle abnormal cases, instead of the common case.

Sure will add it back. I think that catching and ignoring the exceptions here should be revisited in some other change as it seems to me like it could case dataloss that we just assume the exception here means EOF.

Agree. We should investigate that - but let us do it separately from this PR, since this change will be beneficial even without that.

core/src/main/scala/org/apache/spark/util/collection/ExternalAppendOnlyMap.scala

mridulm · 2022-11-02T06:14:50Z

The PR as such looks reasonable to me - can we add a test to explicitly test for EOF behavior ?

+CC @JoshRosen who had worked on this in the distant past :-)
+CC @Ngone51

I want to make sure there are more eyes on this change.

mridulm · 2023-01-10T21:16:55Z

Want to see if we can make this for 3.4 - more eyes on it would be good.

+CC @Ngone51, @srowen, @dongjoon-hyun

dongjoon-hyun · 2023-01-10T21:19:08Z

Thank you for pinging me, @mridulm .
Also, cc @sunchao too.

dongjoon-hyun

Could you rebase to master branch and run the microbenchmarks?

We run the benchmark via GitHub Action. Please see Running benchmarks in your forked repository section of our developer guide, @eejbyfeldt .

https://spark.apache.org/developer-tools.html

eejbyfeldt · 2023-01-11T12:26:53Z

So I ran the benchmark:

================================================================================================
Benchmark Kryo Unsafe vs safe Serialization
================================================================================================

OpenJDK 64-Bit Server VM 1.8.0_352-b08 on Linux 5.15.0-1023-azure
Intel(R) Xeon(R) Platinum 8272CL CPU @ 2.60GHz
Benchmark Kryo Unsafe vs safe Serialization:  Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
---------------------------------------------------------------------------------------------------------------------------
basicTypes: Int with unsafe:true                       224            233           9          4.5         223.9       1.0X
basicTypes: Long with unsafe:true                      253            255           3          4.0         252.7       0.9X
basicTypes: Float with unsafe:true                     237            240           3          4.2         237.1       0.9X
basicTypes: Double with unsafe:true                    237            240           3          4.2         237.1       0.9X
Array: Int with unsafe:true                              4              5           0        235.5           4.2      52.7X
Array: Long with unsafe:true                             7              7           0        149.6           6.7      33.5X
Array: Float with unsafe:true                            4              4           0        247.3           4.0      55.4X
Array: Double with unsafe:true                           7              7           0        143.0           7.0      32.0X
Map of string->Double  with unsafe:true                 41             41           1         24.4          41.0       5.5X
basicTypes: Int with unsafe:false                      257            260           4          3.9         256.7       0.9X
basicTypes: Long with unsafe:false                     279            281           2          3.6         279.1       0.8X
basicTypes: Float with unsafe:false                    252            256           3          4.0         251.9       0.9X
basicTypes: Double with unsafe:false                   260            261           2          3.8         260.0       0.9X
Array: Int with unsafe:false                            24             25           0         41.3          24.2       9.2X
Array: Long with unsafe:false                           33             34           0         30.0          33.3       6.7X
Array: Float with unsafe:false                           9              9           0        109.3           9.2      24.5X
Array: Double with unsafe:false                         16             16           0         63.3          15.8      14.2X
Map of string->Double  with unsafe:false                42             43           1         23.7          42.2       5.3X

This is seems to be within the stdev of what we have in the master branch. Which is expected since this code does not use the interface that that uses the deserialization stream iterators. (And it used the same cpu)

For KryoSerializerBenchmark the branch had:

================================================================================================
Benchmark KryoPool vs old"pool of 1" implementation
================================================================================================

OpenJDK 64-Bit Server VM 1.8.0_352-b08 on Linux 5.15.0-1023-azure
Intel(R) Xeon(R) Platinum 8272CL CPU @ 2.60GHz
Benchmark KryoPool vs old"pool of 1" implementation:  Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
-----------------------------------------------------------------------------------------------------------------------------------
KryoPool:true                                                 8289          10867         NaN          0.0    16577450.0       1.0X
KryoPool:false                                               12592          15035         NaN          0.0    25184133.2       0.7X

This is a slower than what we have on master:

OpenJDK 64-Bit Server VM 1.8.0_352-b08 on Linux 5.15.0-1023-azure
Intel(R) Xeon(R) Platinum 8370C CPU @ 2.80GHz
Benchmark KryoPool vs old"pool of 1" implementation:  Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
-----------------------------------------------------------------------------------------------------------------------------------
KryoPool:true                                                 7098           8972         NaN          0.0    14196810.5       1.0X
KryoPool:false                                               10232          11945         744          0.0    20464754.5       0.7X

But it ran with a different (slower) cpu then master. I also ran the benchmark on current master

OpenJDK 64-Bit Server VM 1.8.0_352-b08 on Linux 5.15.0-1023-azure
Intel(R) Xeon(R) CPU E5-2673 v4 @ 2.30GHz
Benchmark KryoPool vs old"pool of 1" implementation:  Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
-----------------------------------------------------------------------------------------------------------------------------------
KryoPool:true                                                 9646          13191         NaN          0.0    19292739.3       1.0X
KryoPool:false                                               14323          17933         433          0.0    28645212.3       0.7X

There it was even slower since it used an even slower cpu.

Is there someway to better control the cpu used? Or should I just run the benchmark a couple of times?

eejbyfeldt · 2023-01-11T14:03:16Z

I ran the master branch again and used an executor with the same cpu.

OpenJDK 64-Bit Server VM 1.8.0_352-b08 on Linux 5.15.0-1023-azure
Intel(R) Xeon(R) Platinum 8272CL CPU @ 2.60GHz
Benchmark KryoPool vs old"pool of 1" implementation:  Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
-----------------------------------------------------------------------------------------------------------------------------------
KryoPool:true                                                 9375          12171         NaN          0.0    18750400.9       1.0X
KryoPool:false                                               13849          16799         NaN          0.0    27697646.0       0.7X

Based on this it looks like the branch is branch might is a bit faster. But I think it might also be in noise territory and that one would need a more specific benchmark that creates a lot of small streams for the difference to show up. I think it only expected to be order of percent better in the "worst case" when we are creating lots of small streams.

dongjoon-hyun · 2023-01-11T18:11:21Z

Thank you for sharing the result. Without a clear win, it's hard for us to accept this proposal because this is one of the crucial part.

Could you add a benchmark for your specific cases (lots of small streams)?
If there is no regression in the existing benchmarks, your new benchmark can provide us more explicit evidence of this PR's contribution and help us to build a consensus on this direction.

eejbyfeldt · 2023-01-12T13:04:08Z

The PR as such looks reasonable to me - can we add a test to explicitly test for EOF behavior ?

@mridulm I added a spec for this in: 77e616a

Could you add a benchmark for your specific cases (lots of small streams)?

Added a benchmark that shows that there is overhead in using asIterator.toArray compared to just reading the number expected elements in the current master that goes away in this branch. Add results from master with benchmark added (this branch: https://github.com/eejbyfeldt/spark/tree/SPARK-40912-only-adding-benchmark) in 7580633 and then overwrote them in with this branch in bc011c6

srowen · 2023-05-05T15:32:33Z

Looks OK to me; @mridulm ?

dongjoon-hyun

Could you run the benchmark once more? The generated files look wrong to me because the Java version is downgraded.

- OpenJDK 64-Bit Server VM 11.0.18+10 on Linux 5.15.0-1031-azure
- Intel(R) Xeon(R) CPU E5-2673 v3 @ 2.40GHz
+ OpenJDK 64-Bit Server VM 11.0.17+8 on Linux 5.15.0-1031-azure
+ Intel(R) Xeon(R) Platinum 8370C CPU @ 2.80GHz

According to the document, 11.0.18+10 (default) is supposed to be there.

https://github.com/actions/runner-images/blob/main/images/linux/Ubuntu2004-Readme.md

mridulm · 2023-05-06T09:40:35Z

Looks fine to me.
We can merge once @dongjoon-hyun's comment is addressed.

eejbyfeldt · 2023-05-08T06:53:03Z

My plan was to update the benchmarks. But I did not get around to uploading the results until today. But now the branch should be updated with an up to date run.

srowen · 2023-05-10T13:23:54Z

Merged to master

LuciferYang · 2023-05-10T16:15:53Z

core/src/test/scala/org/apache/spark/serializer/KryoIteratorBenchmark.scala

+    val name = "Benchmark of kryo asIterator on deserialization stream"
+    runBenchmark(name) {
+      val benchmark = new Benchmark(name, N, 10, output = output)
+      Seq(true, false).map(useIterator => run(useIterator, benchmark))


nit: should use .foreach instead of .map

LuciferYang · 2023-05-10T16:16:23Z

core/src/test/scala/org/apache/spark/serializer/KryoIteratorBenchmark.scala

+      val elements = Array.fill[T](elementCount)(createElement)
+
+      benchmark.addCase(
+        s"Colletion of $name with $elementCount elements, useIterator: $useIterator") { _ =>


Typo: Colletion -> Collection

LuciferYang · 2023-05-10T16:24:56Z

core/src/test/scala/org/apache/spark/serializer/KryoIteratorBenchmark.scala

+        useIterator: Boolean,
+        ser: SerializerInstance): Int = {
+      val serialized: Array[Byte] = {
+        val baos = new ByteArrayOutputStream()


The initial size of ByteArrayOutputStream is 32. Will the grow of underlying byte[] and GC affect the test results? If so, is it possible to estimate a reasonable initial size?

The GC will/might make the benchmark more noisy but it should not introduce any bias?

I guess choosing a bigger initial size will reduce the issue for some of the benchmark for some of the cases as it will not need to resize, but I can not see any simple way to estimate the total size in general. But maybe using a bigger initial size is better/good enough?

github-actions bot added CORE DSTREAM labels Oct 28, 2022

mridulm reviewed Oct 30, 2022

View reviewed changes

eejbyfeldt force-pushed the SPARK-40912 branch from cff6396 to 017cab8 Compare November 1, 2022 11:54

eejbyfeldt changed the title ~~[SPARK-40912][CORE][WIP] Overhead of Exceptions in DeserializationStream~~ [SPARK-40912][CORE][WIP] Overhead of Exceptions in KryoDeserializationStream Nov 1, 2022

eejbyfeldt force-pushed the SPARK-40912 branch from 93486cd to 208a430 Compare November 1, 2022 12:13

dongjoon-hyun reviewed Jan 10, 2023

View reviewed changes

eejbyfeldt force-pushed the SPARK-40912 branch from 208a430 to 86eda97 Compare January 11, 2023 07:42

github-actions bot removed the DSTREAM label Jan 11, 2023

eejbyfeldt changed the title ~~[SPARK-40912][CORE][WIP] Overhead of Exceptions in KryoDeserializationStream~~ [SPARK-40912][CORE]Overhead of Exceptions in KryoDeserializationStream Jan 11, 2023

eejbyfeldt force-pushed the SPARK-40912 branch from b1253ae to aad5357 Compare January 12, 2023 07:56

eejbyfeldt requested review from mridulm and dongjoon-hyun and removed request for mridulm and dongjoon-hyun January 16, 2023 12:15

eejbyfeldt force-pushed the SPARK-40912 branch from 84eb03b to 76f24da Compare May 3, 2023 09:20

dongjoon-hyun reviewed May 5, 2023

View reviewed changes

eejbyfeldt added 16 commits May 8, 2023 11:20

no exceptions in KryoDeserializationStream

28c032b

Fix spec that depend on handling of broken data

eb4f6a5

Remove dependency on EOFException in ExternalAppendOnlyMap

983a2c3

Use NextIterator

c4a998c

PR comments, make changes more conservative

9d8b3fa

Even more conservative

74dbcae

Add test showing that we still ignore truncated buffers

5934d29

Add benchmark

598be1e

Add results from master branch

ac63591

Add results from this branch

a55bfb2

Also verify readValue throws EOFException

689a1b1

Add one more benchmark case

3992b24

Add result for master version

8b3ee16

Add benchmark results

05fb95b

PR comments

7526c9f

Add updated benchmark runs

a8967d8

eejbyfeldt force-pushed the SPARK-40912 branch from 6ec2043 to a8967d8 Compare May 8, 2023 09:20

srowen approved these changes May 9, 2023

View reviewed changes

srowen closed this in 4def99d May 10, 2023

LuciferYang reviewed May 10, 2023

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-40912][CORE]Overhead of Exceptions in KryoDeserializationStream #38428

[SPARK-40912][CORE]Overhead of Exceptions in KryoDeserializationStream #38428

eejbyfeldt commented Oct 28, 2022

AmplabJenkins commented Oct 28, 2022

mridulm left a comment

mridulm Oct 30, 2022

eejbyfeldt Nov 1, 2022

mridulm Nov 2, 2022 •

edited

mridulm Oct 30, 2022

eejbyfeldt Nov 1, 2022

mridulm Nov 2, 2022

mridulm commented Nov 2, 2022

mridulm commented Jan 10, 2023

dongjoon-hyun commented Jan 10, 2023

dongjoon-hyun left a comment •

edited

eejbyfeldt commented Jan 11, 2023 •

edited

eejbyfeldt commented Jan 11, 2023

dongjoon-hyun commented Jan 11, 2023

eejbyfeldt commented Jan 12, 2023

srowen commented May 5, 2023

dongjoon-hyun left a comment

mridulm commented May 6, 2023

eejbyfeldt commented May 8, 2023

srowen commented May 10, 2023

LuciferYang May 10, 2023

LuciferYang May 10, 2023

LuciferYang May 10, 2023 •

edited

eejbyfeldt May 15, 2023

[SPARK-40912][CORE]Overhead of Exceptions in KryoDeserializationStream #38428

[SPARK-40912][CORE]Overhead of Exceptions in KryoDeserializationStream #38428

Conversation

eejbyfeldt commented Oct 28, 2022

What changes were proposed in this pull request?

Why are the changes needed?

Does this PR introduce any user-facing change?

How was this patch tested?

AmplabJenkins commented Oct 28, 2022

mridulm left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mridulm Nov 2, 2022 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mridulm commented Nov 2, 2022

mridulm commented Jan 10, 2023

dongjoon-hyun commented Jan 10, 2023

dongjoon-hyun left a comment • edited

Choose a reason for hiding this comment

eejbyfeldt commented Jan 11, 2023 • edited

eejbyfeldt commented Jan 11, 2023

dongjoon-hyun commented Jan 11, 2023

eejbyfeldt commented Jan 12, 2023

srowen commented May 5, 2023

dongjoon-hyun left a comment

Choose a reason for hiding this comment

mridulm commented May 6, 2023

eejbyfeldt commented May 8, 2023

srowen commented May 10, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

LuciferYang May 10, 2023 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

mridulm Nov 2, 2022 •

edited

dongjoon-hyun left a comment •

edited

eejbyfeldt commented Jan 11, 2023 •

edited

LuciferYang May 10, 2023 •

edited