[SPARK-25299] Make UnsafeShuffleWriter use the new API by ifilonenko · Pull Request #9 · bloomberg/apache-spark-on-k8s

ifilonenko · 2019-03-28T23:21:03Z

What changes were proposed in this pull request?

Make UnsafeShuffleWriter use the new API

Shuffle Writes [4/6]

How was this patch tested?

Compiled and unit tests

ifilonenko · 2019-03-28T23:21:27Z

@mccheah @yifeih for initial opinions

* But don't close the partition stream returned by the partition writer.

ifilonenko · 2019-04-09T01:19:35Z

@mccheah @yifeih this PR is ready for merge after approval

mccheah · 2019-04-10T17:25:49Z

core/src/main/java/org/apache/spark/shuffle/sort/UnsafeShuffleWriter.java

+        ShufflePartitionWriter writer = null;
+        try {
+          writer = mapWriter.getNextPartitionWriter();
+          writer.toStream();


Do we need to do this?

Without this the file does not get created and it fails UnsafeShuffleWriter.writeEmptyIterator

When we commit all partitions, should we not check that we never created a writer; if we never did then we need to make the file?

Agreed. I believe we talked about this in the first writer PR where the empty file was made by blockResolver.writeIndexFileAndCommit()?

It seems that the test is expecting a file.... idk why, so maybe we should bring that create file method back?

We've always expected a file, that's correct. The question is why do we have to call writer.toStream to ensure the file is created? I'm fairly certain the error is in how we wrote the plugin.

Which is why, i think on commit we might need to create an empty file if a file does not exist. As we initially wrote but decided to take out

We thought that blockIndexResolver.writeIndexFileAndCommit should create the file. Why is it not doing so in this case?

mccheah · 2019-04-10T17:26:38Z

core/src/main/java/org/apache/spark/shuffle/sort/UnsafeShuffleWriter.java

-    final CountingOutputStream mergedFileOutputStream = new CountingOutputStream(bos);
-
    boolean threwException = true;
+


No need for this blank line

mccheah · 2019-04-10T17:28:12Z

core/src/main/java/org/apache/spark/shuffle/sort/UnsafeShuffleWriter.java

-    FileChannel mergedFileOutputChannel = null;

    boolean threwException = true;
+


No need for blank line

mccheah · 2019-04-10T17:28:33Z

core/src/main/java/org/apache/spark/shuffle/sort/UnsafeShuffleWriter.java

+            writeMetrics.incWriteTime(System.nanoTime() - writeStartTime);
+          }
+        } catch (IOException e) {
+          throw e;


Why do we have to throw e here?

mccheah · 2019-04-10T17:29:34Z

core/src/main/java/org/apache/spark/shuffle/sort/io/DefaultShuffleMapOutputWriter.java

      outputFileChannel.close();
    }
-
+    


No need for blank line

mccheah · 2019-04-10T17:30:26Z

core/src/test/java/org/apache/spark/shuffle/sort/UnsafeShuffleWriterSuite.java

+
+    Random rand = new Random();
+    TaskContext$.MODULE$.setTaskContext(new TaskContextImpl(
+      0, 0, 0, rand.nextInt(10000), 0, taskMemoryManager, new Properties(), null, taskMetrics));


Any reason we can't use a fixed number instead of a random one?

yifeih

just a few questions and comments, other than that, looks good to me!

yifeih · 2019-04-10T19:26:04Z

core/src/main/java/org/apache/spark/shuffle/sort/UnsafeShuffleWriter.java

+        ShufflePartitionWriter writer = null;
+        try {
+          writer = mapWriter.getNextPartitionWriter();
+          writer.toStream();


Agreed. I believe we talked about this in the first writer PR where the empty file was made by blockResolver.writeIndexFileAndCommit()?

yifeih · 2019-04-10T20:03:05Z

core/src/main/java/org/apache/spark/shuffle/sort/io/DefaultShuffleMapOutputWriter.java


    @Override
-    public void close() throws IOException {
-      flush();


wait, why did we get rid of this?

On one hand it's inconsequential because the contract is the writer above this never closes this stream - it leaves it up to the partition writer to close this.

On the other hand, this is to preserve an optimization we discovered in UnsafeShuffleWriter that encourages us to minimize the number of times we flush the buffer to the underlying file.

We can get correctness by just flushing the merged buffered file stream at the very end when we commit all the partitions.

yifeih · 2019-04-10T20:05:29Z

core/src/test/java/org/apache/spark/shuffle/sort/UnsafeShuffleWriterSuite.java

+    when(taskContext.taskMemoryManager()).thenReturn(taskMemoryManager);
+
+    Random rand = new Random();
+    TaskContext$.MODULE$.setTaskContext(new TaskContextImpl(


I'm a little confused about this because there's already a taskContext object that's mocked. why can't you just set it to that?

mccheah · 2019-04-15T22:20:52Z

core/src/main/java/org/apache/spark/shuffle/sort/io/DefaultShuffleMapOutputWriter.java

  public void commitAllPartitions() throws IOException {
    cleanUp();
    blockResolver.writeIndexFileAndCommit(shuffleId, mapId, partitionLengths, outputTempFile);
+    if (!outputFile.exists()) {


Why isn't blockResolver#writeIndexFileAndCommit creating the file?

mccheah · 2019-04-15T23:26:43Z

core/src/main/java/org/apache/spark/shuffle/sort/UnsafeShuffleWriter.java

    final boolean encryptionEnabled = blockManager.serializerManager().encryptionEnabled();
+    final int numPartitions = partitioner.numPartitions();
+    long[] partitionLengths = new long[numPartitions];
+    logger.error(mapWriter.toString());


Don't do this

meep, i left a log statement :(

mccheah · 2019-04-15T23:39:32Z

core/src/main/java/org/apache/spark/shuffle/sort/UnsafeShuffleWriter.java

-        // output file would have already been counted as shuffle bytes written.
-        Files.move(spills[0].file, outputFile);
-        return spills[0].partitionLengths;
+        ShufflePartitionWriter writer = null;


Do we not need to commitAllPartitions here? I'd think we wouldn't need to create getNextPartitionWriter either.

mccheah · 2019-04-16T00:24:17Z

Ok this looks fine. Please squash and merge into your branch and we'll cherry-pick to our fork. Thanks!

@ifilonenko

Ported from bloomberg#9. Credits to @ifilonenko!

@ifilonenko

…on-k8s#536) Ported from #9. Credits to @ifilonenko!

@ifilonenko

Ported from bloomberg#9. Credits to @ifilonenko!

ifilonenko force-pushed the shuffle-writer-default-impl branch from 91fea02 to 6f0c44b Compare April 7, 2019 17:19

ifilonenko added 2 commits April 7, 2019 10:25

initial push

67ba987

closing input channel error

3ee7d73

ifilonenko force-pushed the shuffle-writer-default-impl-unsafe branch from c6cd4c7 to 3ee7d73 Compare April 7, 2019 17:26

ifilonenko and others added 4 commits April 8, 2019 00:04

added certain fixes to make all tests except transferTo pass

905b188

small fix to make transferTo properly work

0bdf098

Always close the serializer stream. (#10)

f4f6772

* But don't close the partition stream returned by the partition writer.

fix issues that are blocking tests

a9e0afc

ifilonenko changed the title ~~[SPARK-25299][WIP] Make UnsafeShuffleWriter use the new API~~ [SPARK-25299] Make UnsafeShuffleWriter use the new API Apr 9, 2019

ifilonenko added 2 commits April 8, 2019 18:22

style

6ebb5d8

further style

c808ba3

mccheah reviewed Apr 10, 2019

View reviewed changes

yifeih reviewed Apr 10, 2019

View reviewed changes

resolve comments

f74e53e

mccheah reviewed Apr 15, 2019

View reviewed changes

ifilonenko added 2 commits April 15, 2019 15:38

remove unecessary file creation

f6dfff3

remove .toStream

c5fffd6

mccheah reviewed Apr 15, 2019

View reviewed changes

hot fix to remove log statement

66f3d98

mccheah reviewed Apr 15, 2019

View reviewed changes

fix contract

6267e1b

ifilonenko merged commit 1fe33a9 into shuffle-writer-default-impl Apr 16, 2019

mccheah mentioned this pull request Apr 16, 2019

[SPARK-25299] Make UnsafeShuffleWriter use the new API palantir/spark#536

Merged

bulldozer-bot bot pushed a commit to palantir/spark that referenced this pull request Apr 17, 2019

[SPARK-25299] Make UnsafeShuffleWriter use the new API (#536)

8b021b3

Ported from bloomberg#9. Credits to @ifilonenko!

ifilonenko pushed a commit that referenced this pull request Apr 17, 2019

[SPARK-25299] Make UnsafeShuffleWriter use the new API (apache-spark-…

b0ca599

…on-k8s#536) Ported from #9. Credits to @ifilonenko!

mccheah added a commit to palantir/spark that referenced this pull request Jun 27, 2019

[SPARK-25299] Make UnsafeShuffleWriter use the new API (#536)

d13037f

Ported from bloomberg#9. Credits to @ifilonenko!

		final CountingOutputStream mergedFileOutputStream = new CountingOutputStream(bos);

		boolean threwException = true;

		FileChannel mergedFileOutputChannel = null;

		boolean threwException = true;

Conversation

ifilonenko commented Mar 28, 2019

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

ifilonenko commented Mar 28, 2019

Uh oh!

ifilonenko commented Apr 9, 2019

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!

ifilonenko Apr 10, 2019 • 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!

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!

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

mccheah commented Apr 16, 2019

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

ifilonenko Apr 10, 2019 •

edited

Loading