[SPARK-23253][Core][Shuffle]Only write shuffle temporary index file when there is not an existing one #20422

yaooqinn · 2018-01-29T10:15:33Z

What changes were proposed in this pull request?

Shuffle Index temporay file is used for atomic creating shuffle index file, it is not needed when the index file already exists after another attempts of same task had it done.

How was this patch tested?

exitsting ut

cc @squito

… one

srowen · 2018-01-29T12:08:12Z

Is this just trying to reuse a file that should have been cleaned up after prior failure? If so is that possible as a more direct solution? I wonder if there aren't corner cases here where the file exists and it is still being written to by another process. This could result in corruption. But I am not familiar with this mechanism

jerryshao · 2018-01-29T12:51:53Z

I think it is necessary to add unit test to verify the changes.

SparkQA · 2018-01-29T17:18:24Z

Test build #86768 has finished for PR 20422 at commit 98ea6a7.

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

squito · 2018-01-30T03:27:24Z

thanks for taking a look at this @yaooqinn . To clarify -- there is no bug you are trying to fix here, is there? Its just an optimization? From a quick glance I think the change seems correct ... but also seems like such a minor improvement that I'm not sure I see the value in changing this.

yaooqinn · 2018-01-30T03:39:59Z

thanks guys for reviewing. yes, this is just a minor improvement which I guess code here seem not very logical when I was trying to do some optimizations for my customer's heavy shuffle case. If it's too trivial, I can close it.

jerryshao · 2018-01-30T03:44:19Z

I agree with @squito , unless there's a bug in it, it is risky and unnecessary to change the logic in this critical path.

squito · 2018-01-30T04:00:50Z

Jenkins, ok to test

squito

I actually think this is OK, though I'm still somewhat wary of changing this code for minor gains.

It is already tested in IndexShuffleBlockResolverSuite. Btw, I noticed that the comment here is wrong:

https://github.com/apache/spark/blob/master/core/src/test/scala/org/apache/spark/shuffle/sort/IndexShuffleBlockResolverSuite.scala#L126

It should say "the dataFile should be the new one, since we deleted the dataFile from the first attempt".

It would also be good to expand that test to actually read the index file and make sure it has the right values.

squito · 2018-01-30T04:02:00Z

core/src/main/scala/org/apache/spark/shuffle/IndexShuffleBlockResolver.scala

          }
-          indexTmp.delete()
        } else {
+          val out = new DataOutputStream(new BufferedOutputStream(new FileOutputStream(indexTmp)))


move this below the comment "This is the first successul attempt".

I'd also include a comment about why we write to a temporary file, even though we're always going to rename (because in case the task dies somehow, we'd prefer to not leave a half-written index file in the final location).

SparkQA · 2018-01-30T07:30:19Z

Test build #86792 has finished for PR 20422 at commit 98ea6a7.

This patch fails PySpark unit tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2018-01-30T11:19:39Z

Test build #86806 has finished for PR 20422 at commit 6196770.

This patch fails Spark unit tests.
This patch merges cleanly.
This patch adds no public classes.

yaooqinn · 2018-01-30T14:17:46Z

retest this please

SparkQA · 2018-01-30T18:34:27Z

Test build #86831 has finished for PR 20422 at commit 6196770.

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

SparkQA · 2018-01-31T05:56:00Z

Test build #86858 has finished for PR 20422 at commit 87e6dc0.

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

yaooqinn · 2018-01-31T06:15:18Z

@squito add a test for index file. plz check it again, thanks.

squito · 2018-01-31T19:16:13Z

@jerryshao are you ok with making this change? I think our original comments corssed paths as I was taking a closer look

squito

thanks for the update @yaooqinn . i left some style comments, but also waiting to see @jerryshao 's opinion. Sorry for the back and forth, in any case its always useful to have more people looking at this code critically.

squito · 2018-01-31T19:21:07Z

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

+    val idxName = s"shuffle_${shuffleId}_${mapId}_0.index"
+    val resolver = new IndexShuffleBlockResolver(conf, blockManager)
+
+    val lengths = (1 to 2).map(_ => 8L).toArray


you could do Array.fill(2)(8L)

squito · 2018-01-31T19:28:31Z

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

    assert(firstByte2(0) === 2)
  }
+
+  test("SPARK-23253: index files should be created properly") {


thanks for adding this, but actually I'm not sure this is covering any cases in the previous test, is it? I was thinking of just adding something to read the actual index file, and make sure it had the right values to go with the update to the data file (or no updates in some cases).

you may have added a couple more asserts than the original test -- if so, maybe they can just be added to the original?

+1 we can add check the index file in the original test case.

jiangxb1987 · 2018-02-01T00:02:33Z

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

@@ -123,7 +123,7 @@ class IndexShuffleBlockResolverSuite extends SparkFunSuite with BeforeAndAfterEa
    assert(dataFile.length() === 35)
    assert(!dataTmp2.exists())


not related to the change, but this should be assert(!dataTmp3.exists())

jiangxb1987 · 2018-02-01T03:07:53Z

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

 package org.apache.spark.shuffle.sort

-import java.io.{File, FileInputStream, FileOutputStream}
+import java.io._


nit: this change is not necessary

jiangxb1987 · 2018-02-01T03:18:12Z

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

    assert(firstByte(0) === 0)

+    // The index file should not change
+    val secondBytes = new Array[Byte](8)


nit: should have a better name, perhaps indexSecondValue ?

not sure whether we should change this, just my two cents.

SparkQA · 2018-02-01T06:34:37Z

Test build #86905 has finished for PR 20422 at commit a96f6c4.

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

SparkQA · 2018-02-01T06:45:25Z

Test build #86906 has finished for PR 20422 at commit 246dbca.

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

SparkQA · 2018-02-01T07:14:25Z

Test build #86907 has finished for PR 20422 at commit f3f3627.

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

jerryshao

LGTM.

squito

lgtm

squito · 2018-02-01T04:19:09Z

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

+    } {
+      indexIn.close()
+    }
+    assert(secondValueOffset(7) === 10, "The index file should not change")


minor: here and below, would be more clear if you use DataInputStream.readLong() (no magic 7 offset, and you check the rest of the bytes):

val indexIn = new DataInputStream( newFileInputStream(indexFile)) Utils.tryWithSafeFinally { indexIn.readLong() // first offset is always 0 assert(10 === indexIn.readLong(),"The index file should not change") }

jiangxb1987 · 2018-02-01T18:27:11Z

LGTM

SparkQA · 2018-02-02T05:36:15Z

Test build #86958 has finished for PR 20422 at commit 5677448.

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

SparkQA · 2018-02-02T05:38:42Z

Test build #86959 has finished for PR 20422 at commit a3b85f1.

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

squito · 2018-02-02T15:11:36Z

merged to master.

thanks @yaooqinn for doing and updating the tests too

Only write shuffle temporary index file when there is not an existing…

98ea6a7

… one

squito reviewed Jan 30, 2018

View reviewed changes

add a ut

6196770

modify ut

87e6dc0

squito reviewed Jan 31, 2018

View reviewed changes

jiangxb1987 reviewed Feb 1, 2018

View reviewed changes

yaooqinn added 2 commits February 1, 2018 10:54

merge uts together

a96f6c4

minor

246dbca

jiangxb1987 reviewed Feb 1, 2018

View reviewed changes

comments

f3f3627

jerryshao approved these changes Feb 1, 2018

View reviewed changes

squito approved these changes Feb 1, 2018

View reviewed changes

yaooqinn added 2 commits February 2, 2018 09:51

review

5677448

blank line

a3b85f1

asfgit closed this in dd52681 Feb 2, 2018

yaooqinn deleted the SPARK-23253 branch May 16, 2018 07:06

cloud-fan mentioned this pull request Sep 1, 2018

[SPARK-23243][Core] Fix RDD.repartition() data correctness issue #22112

Closed

		@@ -123,7 +123,7 @@ class IndexShuffleBlockResolverSuite extends SparkFunSuite with BeforeAndAfterEa
		assert(dataFile.length() === 35)
		assert(!dataTmp2.exists())

[SPARK-23253][Core][Shuffle]Only write shuffle temporary index file when there is not an existing one #20422

[SPARK-23253][Core][Shuffle]Only write shuffle temporary index file when there is not an existing one #20422

Uh oh!

Conversation

yaooqinn commented Jan 29, 2018

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

srowen commented Jan 29, 2018

Uh oh!

jerryshao commented Jan 29, 2018

Uh oh!

SparkQA commented Jan 29, 2018

Uh oh!

squito commented Jan 30, 2018

Uh oh!

yaooqinn commented Jan 30, 2018

Uh oh!

jerryshao commented Jan 30, 2018

Uh oh!

squito commented Jan 30, 2018

Uh oh!

squito left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Jan 30, 2018

Uh oh!

SparkQA commented Jan 30, 2018

Uh oh!

yaooqinn commented Jan 30, 2018

Uh oh!

SparkQA commented Jan 30, 2018

Uh oh!

SparkQA commented Jan 31, 2018

Uh oh!

yaooqinn commented Jan 31, 2018

Uh oh!

squito commented Jan 31, 2018

Uh oh!

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

jiangxb1987 Feb 1, 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!

SparkQA commented Feb 1, 2018

Uh oh!

SparkQA commented Feb 1, 2018

Uh oh!

SparkQA commented Feb 1, 2018

Uh oh!

jerryshao left a comment

Choose a reason for hiding this comment

Uh oh!

squito left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jiangxb1987 commented Feb 1, 2018

Uh oh!

SparkQA commented Feb 2, 2018

Uh oh!

SparkQA commented Feb 2, 2018

Uh oh!

jiangxb1987 Feb 1, 2018 •

edited

Loading