SPARK-25299: Add Reader Benchmarks #508

yifeih · 2019-03-05T20:49:40Z

No description provided.

…-reader

mccheah

This gives us pretty high fidelity into the shuffle read performance. Nice work. Left some feedback below.

mccheah · 2019-03-15T23:38:47Z

core/src/test/scala/org/apache/spark/shuffle/sort/BlockStoreShuffleReaderBenchmark.scala

+
+  private var tempDir: File = _
+
+  private val SHUFFLE_ID: Int = 0


Is it necessary to declare types here? There's a lot of places in this class where type declaration isn't necessary.

mccheah · 2019-03-15T23:42:39Z

core/src/test/scala/org/apache/spark/shuffle/sort/BlockStoreShuffleReaderBenchmark.scala

+      dependency = dependency)
+
+    // We cannot mock the TaskContext because it taskMetrics() gets called at every next()
+    // call on the reader, and Mockito will try to log all calls to taskMetrics(), thus OOM-ing


Is this even the case when you add a then behavior on the mock? If so then we're not given much of a choice here huh - I don't think this reads as well when inlined like this - perhaps make this a separate inner class entirely.

Yea, i can make it a separate inner class

mccheah · 2019-03-15T23:43:27Z

core/src/test/scala/org/apache/spark/shuffle/sort/BlockStoreShuffleReaderBenchmark.scala

+    println("Generating test data with num records: " + size)
+    val random = new Random(123)
+    val dataOutput = new FileOutputStream(file)
+    val coutingOutput = new CountingOutputStream(dataOutput)


mccheah · 2019-03-15T23:43:39Z

core/src/test/scala/org/apache/spark/shuffle/sort/BlockStoreShuffleReaderBenchmark.scala

+        serializedOutput.writeValue(x)
+      }}
+    }
+    finally {


Move finally up a line

mccheah · 2019-03-15T23:44:46Z

core/src/test/scala/org/apache/spark/shuffle/sort/BlockStoreShuffleReaderBenchmark.scala

+    aggregationBenchmark.run()
+
+
+    val sorter = Ordering.String


Think it's fine enough to inline this.

mccheah · 2019-03-15T23:45:38Z

core/src/test/scala/org/apache/spark/shuffle/sort/BlockStoreShuffleReaderBenchmark.scala

+  }
+
+  def runWithLargeDataset(): Unit = {
+    val size = TEST_DATA_SIZE


This can be inlined

mccheah · 2019-03-15T23:45:50Z

core/src/test/scala/org/apache/spark/shuffle/sort/BlockStoreShuffleReaderBenchmark.scala

+
+  def runWithLargeDataset(): Unit = {
+    val size = TEST_DATA_SIZE
+    val tempDataFile: File = File.createTempFile("test-data", "", tempDir)


No need to declare type

mccheah · 2019-03-15T23:47:50Z

core/src/test/scala/org/apache/spark/shuffle/sort/BlockStoreShuffleReaderBenchmark.scala

+      assert(numRead == size * NUM_MAPS)
+    }
+    sortingBenchmark.addTimerCase("remote rpc fetch") { timer =>
+      val reader = setupReader(


Similar to the exercise we did for writers, let's try to factor out more common logic to helper methods. No need for anything fancy, just reducing code duplication.

core/src/test/scala/org/apache/spark/shuffle/sort/BlockStoreShuffleReaderBenchmark.scala

mccheah · 2019-03-20T03:59:51Z

core/src/test/scala/org/apache/spark/shuffle/sort/BlockStoreShuffleReaderBenchmark.scala

+  def generateDataOnDisk(size: Int, file: File, recordOffset: Int): (Long, Long) = {
+    // scalastyle:off println
+    println("Generating test data with num records: " + size)
+    class ManualCloseFileOutputStream(file: File) extends FileOutputStream(file, true) {


Let's move this out of the method def, a tiny bit easier to read. Bottom of the class, perhaps.

mccheah · 2019-03-20T04:01:08Z

core/src/test/scala/org/apache/spark/shuffle/sort/BlockStoreShuffleReaderBenchmark.scala

+      timer.startTiming()
+      val numRead = reader.read().length
+      timer.stopTiming()
+      if (assertSize.isDefined) {


Nit: Just do assertSize.foreach

mccheah · 2019-03-20T04:10:32Z

core/src/test/scala/org/apache/spark/shuffle/sort/BlockStoreShuffleReaderBenchmark.scala

+  def addBenchmarkCase(
+      benchmark: Benchmark,
+      name: String,
+      func: () => BlockStoreShuffleReader[String, String],


func can be shuffleReaderSupplier to indicate that this is just a factory.

What happens if we make this the following type?

def addBenchmarkCase( benchmark: Benchmark, name: String, shuffleReaderSupplier: => BlockStoreShuffleReader[String, String], // Notice there's no () here assertSize: Option[Int] = None): Unit = {... }

But don't know my Scala offhand to know if this works, or if that changes the way one invokes such a method.

The hope here is that instead of passing in () => setupReader(...), you can instead pass in setupReader(), and we preserve the behavior of the reader being lazily instantiated.

ah it worked! nice :)

mccheah · 2019-03-20T04:10:47Z

core/src/test/scala/org/apache/spark/shuffle/sort/BlockStoreShuffleReaderBenchmark.scala

+      benchmark: Benchmark,
+      name: String,
+      func: () => BlockStoreShuffleReader[String, String],
+      assertSize: Option[Int] = Option.empty): Unit = {


Default to None instead - a little more concise.

svc-spark-25299

================================================================================================
BlockStoreShuffleReader reader
================================================================================================

Java HotSpot(TM) 64-Bit Server VM 1.8.0_131-b11 on Linux 4.15.0-1014-gcp
Intel(R) Xeon(R) CPU @ 2.30GHz
no aggregation or sorting:                Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
------------------------------------------------------------------------------------------------------------------------
local fetch                                       10614          10773         141          0.9        1061.4       1.0X
remote rpc fetch                                  10632          10686          38          0.9        1063.2       1.0X

Java HotSpot(TM) 64-Bit Server VM 1.8.0_131-b11 on Linux 4.15.0-1014-gcp
Intel(R) Xeon(R) CPU @ 2.30GHz
with aggregation:                         Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
------------------------------------------------------------------------------------------------------------------------
local fetch                                       29734          30320         440          0.1       14867.0       1.0X
remote rpc fetch                                  29787          30162         247          0.1       14893.3       1.0X

Java HotSpot(TM) 64-Bit Server VM 1.8.0_131-b11 on Linux 4.15.0-1014-gcp
Intel(R) Xeon(R) CPU @ 2.30GHz
with sorting:                             Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
------------------------------------------------------------------------------------------------------------------------
local fetch                                       31650          32065         563          0.1       15825.0       1.0X
remote rpc fetch                                  31713          32201         294          0.1       15856.7       1.0X

Java HotSpot(TM) 64-Bit Server VM 1.8.0_131-b11 on Linux 4.15.0-1014-gcp
Intel(R) Xeon(R) CPU @ 2.30GHz
with seek:                                Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
------------------------------------------------------------------------------------------------------------------------
seek to last record                                   1              1           1       3480.0           0.3       1.0X

================================================================================================
BypassMergeSortShuffleWriter write
================================================================================================

Java HotSpot(TM) 64-Bit Server VM 1.8.0_131-b11 on Linux 4.15.0-1014-gcp
Intel(R) Xeon(R) CPU @ 2.30GHz
BypassMergeSortShuffleWrite without spill:  Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
------------------------------------------------------------------------------------------------------------------------
small dataset without disk spill                      2              3           2          0.5        2075.4       1.0X

Java HotSpot(TM) 64-Bit Server VM 1.8.0_131-b11 on Linux 4.15.0-1014-gcp
Intel(R) Xeon(R) CPU @ 2.30GHz
BypassMergeSortShuffleWrite with spill:   Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
------------------------------------------------------------------------------------------------------------------------
without transferTo                                 7460           7540          43          0.9        1111.6       1.0X
with transferTo                                    7475           7524          29          0.9        1113.9       1.0X

================================================================================================
SortShuffleWriter writer
================================================================================================

Java HotSpot(TM) 64-Bit Server VM 1.8.0_131-b11 on Linux 4.15.0-1014-gcp
Intel(R) Xeon(R) CPU @ 2.30GHz
SortShuffleWriter without spills:         Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
------------------------------------------------------------------------------------------------------------------------
small dataset without spills                         11             15           4          0.1       10576.8       1.0X

Java HotSpot(TM) 64-Bit Server VM 1.8.0_131-b11 on Linux 4.15.0-1014-gcp
Intel(R) Xeon(R) CPU @ 2.30GHz
SortShuffleWriter with spills:            Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
------------------------------------------------------------------------------------------------------------------------
no map side combine                               14029          14144          92          0.5        2090.4       1.0X
with map side aggregation                         13997          14157         182          0.5        2085.8       1.0X
with map side sort                                13988          14093          60          0.5        2084.4       1.0X

================================================================================================
UnsafeShuffleWriter write
================================================================================================

Java HotSpot(TM) 64-Bit Server VM 1.8.0_131-b11 on Linux 4.15.0-1014-gcp
Intel(R) Xeon(R) CPU @ 2.30GHz
UnsafeShuffleWriter without spills:       Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
------------------------------------------------------------------------------------------------------------------------
small dataset without spills                         21             25           4          0.0       21124.1       1.0X

Java HotSpot(TM) 64-Bit Server VM 1.8.0_131-b11 on Linux 4.15.0-1014-gcp
Intel(R) Xeon(R) CPU @ 2.30GHz
UnsafeShuffleWriter with spills:          Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
------------------------------------------------------------------------------------------------------------------------
without transferTo                                15837          15997         145          0.8        1180.0       1.0X
with transferTo                                   15852          15943          78          0.8        1181.1       1.0X

yifeih added 30 commits February 22, 2019 17:42

add initial bypass merge sort shuffle writer benchmarks

c7abec6

dd unsafe shuffle writer benchmarks

22ef648

changes in bypassmergesort benchmarks

4084e27

cleanup

fb8266d

add circle script

89104e2

add this branch for testing

b90b381

fix circle attempt 1

5e13dd8

checkout code

845e645

add some caches?

a68f459

why is it not pull caches...

757f6fe

save as artifact instead of publishing

0bcd5d9

mkdir

26c01ec

typo

0d7a036

try uploading artifacts again

3fc5331

try print per iteration to avoid circle erroring out on idle

8c33701

blah (#495)

9546397

make a PR comment

d72ba73

actually delete files

1859805

run benchmarks on test build branch

c20f0be

oops forgot to enable upload

444d46a

add sort shuffle writer benchmarks

2322933

add stdev

da0d91c

cleanup sort a bit

e590917

fix stdev text

cbfdb99

fix sort shuffle

cbe38c6

initial code for read side

acdda71

format

fd7a7c5

use times and sample stdev

d82618b

add assert for at least one iteration

610ea1d

cleanup shuffle write to use fewer mocks and single base interface

295d7f3

yifeih added 5 commits March 6, 2019 12:00

try machine execution

33a1b72

try uploading benchmarks (#498)

fa1b96c

Merge branch 'yh/add-benchmarks-and-ci' into yh/add-benchmarks-and-ci…

f02a206

…-reader

don't upload results yet

5675b01

only upload results when merging into the feature branch

37cef1f

yifeih mentioned this pull request Mar 11, 2019

SPARK-25299: add CI infrastructure and SortShuffleWriterBenchmark #498

Merged

yifeih added 4 commits March 12, 2019 15:35

lock down machine image

459e1b5

don't write input data to disk

4cabdbd

Merge branch 'yh/add-benchmarks-and-ci' into yh/add-benchmarks-and-ci…

9e329b4

…-reader

Merge branch 'spark-25299' into yh/add-benchmarks-and-ci-reader

1658370

yifeih changed the base branch from yh/add-benchmarks-and-ci to spark-25299 March 14, 2019 20:58

mccheah requested changes Mar 15, 2019

View reviewed changes

yifeih added 5 commits March 18, 2019 10:52

address some comments

62b9f8b

reader wip

096d981

works

9868a50

refactor reader

1f05a7e

delete unnecessary close/flushes

4ce93c1

mccheah reviewed Mar 20, 2019

View reviewed changes

yifeih added 3 commits March 19, 2019 23:20

comments

34653f2

get rid of some typedefs

c105305

move to bottom of class

5701de1

mccheah approved these changes Mar 26, 2019

View reviewed changes

Merge branch 'spark-25299' into yh/add-benchmarks-and-ci-reader

a3ecd99

yifeih removed the do not merge label Mar 28, 2019

bulldozer-bot bot merged commit adc1918 into spark-25299 Mar 28, 2019

bulldozer-bot bot deleted the yh/add-benchmarks-and-ci-reader branch March 28, 2019 00:57

svc-spark-25299 reviewed Mar 28, 2019

View reviewed changes


		private var tempDir: File = _

		private val SHUFFLE_ID: Int = 0

SPARK-25299: Add Reader Benchmarks #508

SPARK-25299: Add Reader Benchmarks #508

Uh oh!

Conversation

yifeih commented Mar 5, 2019

Uh oh!

mccheah left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

svc-spark-25299 left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!