[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

    // 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

      // 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

      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

    // 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

      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


    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

      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

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

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

Uh oh!

Conversation

caneGuy commented Jan 17, 2018

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

srowen 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!

caneGuy Jan 18, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

caneGuy commented Jan 22, 2018

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!

jerryshao commented Jan 24, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cloud-fan commented Jan 24, 2018

Uh oh!

SparkQA commented Jan 24, 2018

Uh oh!

SparkQA commented Jan 25, 2018

Uh oh!

jerryshao commented Jan 25, 2018

Uh oh!

jiangxb1987 commented Jan 25, 2018

Uh oh!

cloud-fan commented Jan 25, 2018

Uh oh!

caneGuy commented Jan 25, 2018

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

6 participants

caneGuy Jan 18, 2018 •

edited

Loading