[SPARK-18546][core] Fix merging shuffle spills when using encryption. #15982

vanzin · 2016-11-22T21:28:18Z

The problem exists because it's not possible to just concatenate encrypted
partition data from different spill files; currently each partition would
have its own initial vector to set up encryption, and the final merged file
should contain a single initial vector for each merged partiton, otherwise
iterating over each record becomes really hard.

To fix that, UnsafeShuffleWriter now decrypts the partitions when merging,
so that the merged file contains a single initial vector at the start of
the partition data.

Because it's not possible to do that using the fast transferTo path, when
encryption is enabled UnsafeShuffleWriter will revert back to using file
streams when merging. It may be possible to use a hybrid approach when
using encryption, using an intermediate direct buffer when reading from
files and encrypting the data, but that's better left for a separate patch.

As part of the change I made DiskBlockObjectWriter take a SerializerManager
instead of a "wrap stream" closure, since that makes it easier to test the
code without having to mock SerializerManager functionality.

Tested with newly added unit tests (UnsafeShuffleWriterSuite for the write
side and ExternalAppendOnlyMapSuite for integration), and by running some
apps that failed without the fix.

vanzin · 2016-11-22T21:29:02Z

~~This also contains the changes in #15981 since it's built on top of those, but they should be merged separately. The fix for this bug is in the second commit.~~

EDIT: rebased so now diff only contains the fix for this bug.

vanzin · 2016-11-22T21:29:14Z

/cc @zsxwing @JoshRosen

SparkQA · 2016-11-22T21:40:31Z

Test build #69030 has finished for PR 15982 at commit fbd2bca.

This patch fails to build.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2016-11-22T21:49:58Z

Test build #69032 has finished for PR 15982 at commit 3b879f4.

This patch fails build dependency tests.
This patch merges cleanly.
This patch adds no public classes.

vanzin · 2016-11-22T22:34:55Z

retest this please

SparkQA · 2016-11-23T00:54:20Z

Test build #69035 has finished for PR 15982 at commit 3b879f4.

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

vanzin · 2016-11-23T01:11:17Z

Unrelated failure. retest this please

SparkQA · 2016-11-23T03:59:13Z

Test build #69040 has finished for PR 15982 at commit 3b879f4.

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

vanzin · 2016-11-28T17:01:54Z

Ping.

The problem exists because it's not possible to just concatenate encrypted partition data from different spill files; currently each partition would have its own initial vector to set up encryption, and the final merged file should contain a single initial vector for each merged partiton, otherwise iterating over each record becomes really hard. To fix that, UnsafeShuffleWriter now decrypts the partitions when merging, so that the merged file contains a single initial vector at the start of the partition data. Because it's not possible to do that using the fast transferTo path, when encryption is enabled UnsafeShuffleWriter will revert back to using file streams when merging. It may be possible to use a hybrid approach when using encryption, using an intermediate direct buffer when reading from files and encrypting the data, but that's better left for a separate patch. As part of the change I made DiskBlockObjectWriter take a SerializerManager instead of a "wrap stream" closure, since that makes it easier to test the code without having to mock SerializerManager functionality. Tested with newly added unit tests (UnsafeShuffleWriterSuite for the write side and ExternalAppendOnlyMapSuite for integration), and by running some apps that failed without the fix.

zsxwing

Overall looks good. Just left some minor comments

zsxwing · 2016-11-29T18:42:42Z

core/src/main/scala/org/apache/spark/serializer/SerializerManager.scala

    defaultSerializer: Serializer,
    conf: SparkConf,
-    encryptionKey: Option[Array[Byte]]) {
+    val encryptionKey: Option[Array[Byte]]) {


nit: I prefer to add a method called isEncryptionEnabled instead of exposing this field.

zsxwing · 2016-11-29T19:01:30Z

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

    final long[] partitionLengths = new long[numPartitions];
    final InputStream[] spillInputStreams = new FileInputStream[spills.length];
-    OutputStream mergedFileOutputStream = null;
+    final CountingOutputStream mergedFileOutputStream = new CountingOutputStream(


Could you add a comment about why need to use CountingOutputStream + CloseShieldOutputStream? It took me a while to figure out the optimization you did.

zsxwing · 2016-11-29T19:04:52Z

core/src/main/scala/org/apache/spark/serializer/SerializerManager.scala

   * Wrap an output stream for compression if block compression is enabled for its block type
   */
-  private[this] def wrapForCompression(blockId: BlockId, s: OutputStream): OutputStream = {
+  def wrapForCompression(blockId: BlockId, s: OutputStream): OutputStream = {


nit: not need to open this method

zsxwing · 2016-11-29T19:04:56Z

core/src/main/scala/org/apache/spark/serializer/SerializerManager.scala

   * Wrap an input stream for compression if block compression is enabled for its block type
   */
-  private[this] def wrapForCompression(blockId: BlockId, s: InputStream): InputStream = {
+  def wrapForCompression(blockId: BlockId, s: InputStream): InputStream = {


nit: not need to open this method

zsxwing · 2016-11-29T19:12:48Z

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

-              Closeables.close(partitionInputStream, innerThrewException);
+            InputStream partitionInputStream = new LimitedInputStream(spillInputStreams[i],
+              partitionLengthInSpill, false);
+            partitionInputStream = blockManager.serializerManager().wrapForEncryption(


partitionInputStream is not closed

zsxwing · 2016-11-29T19:16:38Z

core/src/test/scala/org/apache/spark/util/collection/ExternalAppendOnlyMapSuite.scala


 package org.apache.spark.util.collection

+import java.security.PrivilegedExceptionAction


nit: unused import

zsxwing · 2016-11-29T19:16:42Z

core/src/test/scala/org/apache/spark/util/collection/ExternalAppendOnlyMapSuite.scala

 import scala.collection.mutable.ArrayBuffer

 import org.apache.spark._
+import org.apache.spark.deploy.SparkHadoopUtil


nit: unused import

zsxwing · 2016-11-29T19:26:30Z

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

+  }
+
+  @Test
+  public void mergeSpillsWithCompressionAndEncryptionSlowPath() throws Exception {


We should also test testMergingSpills(false, null, true); and testMergingSpills(true, null, true).

SparkQA · 2016-11-29T20:19:30Z

Test build #69337 has finished for PR 15982 at commit 2e03ee6.

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

squito · 2016-11-29T21:47:39Z

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

+          partitionOutput = compressionCodec.compressedOutputStream(partitionOutput);
        }
-
+        partitionOutput = new TimeTrackingOutputStream(writeMetrics, partitionOutput);


another change here is that TimeTrackingOutputStream now goes around the compression codec. I think that is the right change, but its at least worth mentioning in the commit msg.

I'm wondering if this its worth having a separate jira for this, just since it will effect metrics for all users

Hmm... let me revert this and open a bug. DiskBlockObjectWriter doesn't count the time for compression / encryption, so this should behave the same. Both should be fixed together.

squito · 2016-11-29T22:13:36Z

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

 import org.apache.spark.ShuffleDependency;
 import org.apache.spark.SparkConf;
 import org.apache.spark.TaskContext;
+import org.apache.spark.deploy.SparkHadoopUtil;


other than CryptoStreamUtils, the other added imports look unused. Also looks like you can eliminate AbstractFunction1 and ByteStreams since you are no longer using them.

squito · 2016-11-29T22:17:55Z

core/src/test/java/org/apache/spark/unsafe/map/AbstractBytesToBytesMapSuite.java

  @Mock(answer = RETURNS_SMART_NULLS) BlockManager blockManager;
  @Mock(answer = RETURNS_SMART_NULLS) DiskBlockManager diskBlockManager;

-  private static final class WrapStream extends AbstractFunction1<OutputStream, OutputStream> {


you can eliminate the imports of AbstractFunction1 and OutputStream after this

squito · 2016-11-29T22:18:27Z

core/src/test/java/org/apache/spark/util/collection/unsafe/sort/UnsafeExternalSorterSuite.java


-  private final long pageSizeBytes = new SparkConf().getSizeAsBytes("spark.buffer.pageSize", "4m");
-
-  private static final class WrapStream extends AbstractFunction1<OutputStream, OutputStream> {


same on trimming imports

SparkQA · 2016-11-29T23:43:50Z

Test build #69348 has finished for PR 15982 at commit 8ac9276.

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

SparkQA · 2016-11-30T01:08:19Z

Test build #69360 has finished for PR 15982 at commit 1025c6b.

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

vanzin · 2016-11-30T02:32:52Z

retest this please

SparkQA · 2016-11-30T05:13:35Z

Test build #69375 has finished for PR 15982 at commit 1025c6b.

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

squito · 2016-11-30T17:44:00Z

I think the comment on #mergeSpillsWithFileStreams needs to be updated slightly to include encryption, but other than that lgtm.

zsxwing · 2016-11-30T18:34:10Z

LGTM

SparkQA · 2016-11-30T21:06:55Z

Test build #69419 has finished for PR 15982 at commit 49737d9.

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

vanzin · 2016-11-30T22:10:03Z

Merging to master / 2.1.

The problem exists because it's not possible to just concatenate encrypted partition data from different spill files; currently each partition would have its own initial vector to set up encryption, and the final merged file should contain a single initial vector for each merged partiton, otherwise iterating over each record becomes really hard. To fix that, UnsafeShuffleWriter now decrypts the partitions when merging, so that the merged file contains a single initial vector at the start of the partition data. Because it's not possible to do that using the fast transferTo path, when encryption is enabled UnsafeShuffleWriter will revert back to using file streams when merging. It may be possible to use a hybrid approach when using encryption, using an intermediate direct buffer when reading from files and encrypting the data, but that's better left for a separate patch. As part of the change I made DiskBlockObjectWriter take a SerializerManager instead of a "wrap stream" closure, since that makes it easier to test the code without having to mock SerializerManager functionality. Tested with newly added unit tests (UnsafeShuffleWriterSuite for the write side and ExternalAppendOnlyMapSuite for integration), and by running some apps that failed without the fix. Author: Marcelo Vanzin <[email protected]> Closes #15982 from vanzin/SPARK-18546. (cherry picked from commit 93e9d88) Signed-off-by: Marcelo Vanzin <[email protected]>

The problem exists because it's not possible to just concatenate encrypted partition data from different spill files; currently each partition would have its own initial vector to set up encryption, and the final merged file should contain a single initial vector for each merged partiton, otherwise iterating over each record becomes really hard. To fix that, UnsafeShuffleWriter now decrypts the partitions when merging, so that the merged file contains a single initial vector at the start of the partition data. Because it's not possible to do that using the fast transferTo path, when encryption is enabled UnsafeShuffleWriter will revert back to using file streams when merging. It may be possible to use a hybrid approach when using encryption, using an intermediate direct buffer when reading from files and encrypting the data, but that's better left for a separate patch. As part of the change I made DiskBlockObjectWriter take a SerializerManager instead of a "wrap stream" closure, since that makes it easier to test the code without having to mock SerializerManager functionality. Tested with newly added unit tests (UnsafeShuffleWriterSuite for the write side and ExternalAppendOnlyMapSuite for integration), and by running some apps that failed without the fix. Author: Marcelo Vanzin <[email protected]> Closes apache#15982 from vanzin/SPARK-18546.

vanzin force-pushed the SPARK-18546 branch from fbd2bca to 3b879f4 Compare November 22, 2016 21:42

vanzin force-pushed the SPARK-18546 branch from 3b879f4 to 2e03ee6 Compare November 29, 2016 17:42

zsxwing requested changes Nov 29, 2016

View reviewed changes

Marcelo Vanzin added 2 commits November 29, 2016 12:52

Review feedback.

e502790

Add explicit "encryptionEnabled" method.

8ac9276

squito reviewed Nov 29, 2016

View reviewed changes

More feedback.

1025c6b

Update comment.

49737d9

asfgit closed this in 93e9d88 Nov 30, 2016

vanzin deleted the SPARK-18546 branch November 30, 2016 22:54


		package org.apache.spark.util.collection

		import java.security.PrivilegedExceptionAction


		private final long pageSizeBytes = new SparkConf().getSizeAsBytes("spark.buffer.pageSize", "4m");

		private static final class WrapStream extends AbstractFunction1<OutputStream, OutputStream> {

[SPARK-18546][core] Fix merging shuffle spills when using encryption. #15982

[SPARK-18546][core] Fix merging shuffle spills when using encryption. #15982

Uh oh!

Conversation

vanzin commented Nov 22, 2016

Uh oh!

vanzin commented Nov 22, 2016 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

vanzin commented Nov 22, 2016

Uh oh!

SparkQA commented Nov 22, 2016

Uh oh!

SparkQA commented Nov 22, 2016

Uh oh!

vanzin commented Nov 22, 2016

Uh oh!

SparkQA commented Nov 23, 2016

Uh oh!

vanzin commented Nov 23, 2016

Uh oh!

SparkQA commented Nov 23, 2016

Uh oh!

vanzin commented Nov 28, 2016

Uh oh!

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

SparkQA commented Nov 29, 2016

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!

SparkQA commented Nov 29, 2016

Uh oh!

SparkQA commented Nov 30, 2016

Uh oh!

vanzin commented Nov 30, 2016

Uh oh!

SparkQA commented Nov 30, 2016

Uh oh!

squito commented Nov 30, 2016

Uh oh!

zsxwing commented Nov 30, 2016

Uh oh!

SparkQA commented Nov 30, 2016

Uh oh!

vanzin commented Nov 30, 2016

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

vanzin commented Nov 22, 2016 •

edited

Loading