[SPARK-23129][CORE] Make deserializeStream of DiskMapIterator init lazily #20292

caneGuy · 2018-01-17T09:36:19Z

What changes were proposed in this pull request?

Currently,the deserializeStream in ExternalAppendOnlyMap#DiskMapIterator init when DiskMapIterator instance created.This will cause memory use overhead when ExternalAppendOnlyMap spill too much times.

We can avoid this by making deserializeStream init when it is used the first time.
This patch make deserializeStream init lazily.

How was this patch tested?

Exist tests

srowen

Without even commenting on the logic, a lot about this change is wrong. I'd review it carefully first.

srowen · 2018-01-17T13:49:14Z

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

@@ -463,21 +463,21 @@ class ExternalAppendOnlyMap[K, V, C](

    // An intermediate stream that reads from exactly one batch
    // This guards against pre-fetching and other arbitrary behavior of higher level streams
-    private var deserializeStream = nextBatchStream()
+    private var deserializeStream = null.asInstanceOf[Option[DeserializationStream]]


Much better: ... deserializeStream: Option[DeserializationStream] = None. It's not just the cast, but the fact that you're assigning null to an Option

srowen · 2018-01-17T13:50:23Z

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

      // Note that batchOffsets.length = numBatches + 1 since we did a scan above; check whether
      // we're still in a valid batch.
-      if (batchIndex < batchOffsets.length - 1) {
+      if (batchIndex < batchOffsets.length - 1 && deserializeStream.isDefined) {
        if (deserializeStream != null) {


This check doesn't make sense then. You probably want a deserializeStream.foreach { ... } construct anyway

srowen · 2018-01-17T13:50:44Z

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

-      if (ds != null) {
-        ds.close()
-        deserializeStream = null
+      if (ds != null && ds.isEmpty) {


This check is wrong

jiangxb1987 · 2018-01-17T19:18:00Z

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

@@ -509,8 +509,8 @@ class ExternalAppendOnlyMap[K, V, C](
     */
    private def readNextItem(): (K, C) = {
      try {
-        val k = deserializeStream.readKey().asInstanceOf[K]
-        val c = deserializeStream.readValue().asInstanceOf[C]
+        val k = deserializeStream.get.readKey().asInstanceOf[K]


If developer call next() without checking hasNext then this will trigger clean up, which is not expected.

@jiangxb1987 I think this does not change the original semantic,since it is only call cleanup when EOFException was thrown.Maybe i missed something?

Here may throw NoSuchElementException.I will fix this.Thanks @jiangxb1987

caneGuy · 2018-01-22T05:19:17Z

Ping @jiangxb1987 could you help review this?thanks too much!

jiangxb1987 · 2018-01-22T21:33:54Z

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

@@ -463,21 +463,21 @@ class ExternalAppendOnlyMap[K, V, C](

    // An intermediate stream that reads from exactly one batch
    // This guards against pre-fetching and other arbitrary behavior of higher level streams
-    private var deserializeStream = nextBatchStream()
+    private var deserializeStream: Option[DeserializationStream] = None


Perhaps private var deserializeStream: Option[DeserializationStream] = nextBatchStream() ? Then we can avoid many other code changes.

Thanks @jiangxb1987 But this may make deserializeStream still init when spilling and generate DiskMapIterator

can't we just use null as initial value? For performance critical places we don't need follow Scala style and use None.

This can also make this patch much smaller.

Ok,i will fix later @jiangxb1987 @cloud-fan Thanks for your precious time

jiangxb1987 · 2018-01-23T05:47:39Z

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

@@ -509,8 +509,8 @@ class ExternalAppendOnlyMap[K, V, C](
     */
    private def readNextItem(): (K, C) = {
      try {
-        val k = deserializeStream.readKey().asInstanceOf[K]
-        val c = deserializeStream.readValue().asInstanceOf[C]
+        val k = deserializeStream.get.readKey().asInstanceOf[K]


We should still check deserializeStream is not emtpy here to be safe

jiangxb1987 · 2018-01-23T05:49:14Z

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

@@ -547,9 +555,9 @@ class ExternalAppendOnlyMap[K, V, C](
    private def cleanup() {
      batchIndex = batchOffsets.length  // Prevent reading any other batch
      val ds = deserializeStream


The new val is not actually needed.

jerryshao · 2018-01-24T09:33:11Z

ok to test.

cloud-fan · 2018-01-24T10:40:35Z

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

+          deserializeStream = nextBatchStream()
+          if (deserializeStream == null) {
+            return false
+          }
        }
        nextItem = readNextItem()
      }
      nextItem != null
    }

    override def next(): (K, C) = {


nit:

if (!hasNext) { throw new NoSuchElementException } val item = nextItem nextItem = null item

cloud-fan · 2018-01-24T10:41:10Z

LGTM

SparkQA · 2018-01-24T13:32:09Z

Test build #86582 has finished for PR 20292 at commit 3087208.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-01-25T06:40:07Z

Test build #86618 has finished for PR 20292 at commit a443531.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

jerryshao · 2018-01-25T06:50:37Z

LGTM.

jiangxb1987 · 2018-01-25T06:55:51Z

LGTM

…zily ## What changes were proposed in this pull request? Currently,the deserializeStream in ExternalAppendOnlyMap#DiskMapIterator init when DiskMapIterator instance created.This will cause memory use overhead when ExternalAppendOnlyMap spill too much times. We can avoid this by making deserializeStream init when it is used the first time. This patch make deserializeStream init lazily. ## How was this patch tested? Exist tests Author: zhoukang <[email protected]> Closes #20292 from caneGuy/zhoukang/lay-diskmapiterator. (cherry picked from commit 45b4bbf) Signed-off-by: Wenchen Fan <[email protected]>

cloud-fan · 2018-01-25T07:26:21Z

thanks, merging to master/2.3!

caneGuy · 2018-01-25T08:08:32Z

Thanks for your time! @cloud-fan @jerryshao @jiangxb1987

[SPARK][CORE] Make deserializeStream of DiskMapIterator init lazily

d2bbbe1

srowen requested changes Jan 17, 2018

View reviewed changes

jiangxb1987 reviewed Jan 17, 2018

View reviewed changes

caneGuy added 2 commits January 18, 2018 10:24

Refine comment

bdb32a1

Refine comment

d109cce

jiangxb1987 reviewed Jan 22, 2018

View reviewed changes

jiangxb1987 reviewed Jan 23, 2018

View reviewed changes

Refine comment

3087208

cloud-fan reviewed Jan 24, 2018

View reviewed changes

Refine comment

a443531

asfgit closed this in 45b4bbf Jan 25, 2018

caneGuy deleted the zhoukang/lay-diskmapiterator branch January 26, 2018 02:37

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-23129][CORE] Make deserializeStream of DiskMapIterator init lazily #20292

[SPARK-23129][CORE] Make deserializeStream of DiskMapIterator init lazily #20292

caneGuy commented Jan 17, 2018

srowen left a comment

srowen Jan 17, 2018

srowen Jan 17, 2018

srowen Jan 17, 2018

jiangxb1987 Jan 17, 2018

caneGuy Jan 18, 2018 •

edited

Loading

caneGuy Jan 18, 2018

caneGuy commented Jan 22, 2018

jiangxb1987 Jan 22, 2018

caneGuy Jan 23, 2018

cloud-fan Jan 23, 2018

cloud-fan Jan 23, 2018

caneGuy Jan 23, 2018

jiangxb1987 Jan 23, 2018

jiangxb1987 Jan 23, 2018

jerryshao commented Jan 24, 2018

cloud-fan Jan 24, 2018

cloud-fan commented Jan 24, 2018

SparkQA commented Jan 24, 2018

SparkQA commented Jan 25, 2018

jerryshao commented Jan 25, 2018

jiangxb1987 commented Jan 25, 2018

cloud-fan commented Jan 25, 2018

caneGuy commented Jan 25, 2018

[SPARK-23129][CORE] Make deserializeStream of DiskMapIterator init lazily #20292

[SPARK-23129][CORE] Make deserializeStream of DiskMapIterator init lazily #20292

Conversation

caneGuy commented Jan 17, 2018

What changes were proposed in this pull request?

How was this patch tested?

srowen left a comment

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

Choose a reason for hiding this comment

caneGuy Jan 18, 2018 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

caneGuy commented Jan 22, 2018

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jerryshao commented Jan 24, 2018

Choose a reason for hiding this comment

cloud-fan commented Jan 24, 2018

SparkQA commented Jan 24, 2018

SparkQA commented Jan 25, 2018

jerryshao commented Jan 25, 2018

jiangxb1987 commented Jan 25, 2018

cloud-fan commented Jan 25, 2018

caneGuy commented Jan 25, 2018

caneGuy Jan 18, 2018 •

edited

Loading