[SPARK-20994] Remove redundant characters in OpenBlocks to save memory for shuffle service. #18231

jinxing64 · 2017-06-07T09:51:00Z

What changes were proposed in this pull request?

In current code, blockIds in OpenBlocks are stored in the iterator on shuffle service.
There are some redundant characters in blockId("shuffle_" + shuffleId + "_" + mapId + "_" + reduceId). This pr proposes to improve the footprint and alleviate the memory pressure on shuffle service.

…for shuffle service.

jinxing64 · 2017-06-07T09:51:30Z

n my cluster, we are suffering from OOM of shuffle-service.
We found that a lot of executors are fetching blocks from a single shuffle-service. Analyzing the memory, we found that the blockIds(shuffle_shuffleId_mapId_reduceId) takes about 1.5GBytes.

srowen

Can that really save much memory? Seems trivial

SparkQA · 2017-06-07T12:01:45Z

Test build #77795 has finished for PR 18231 at commit 96d07aa.

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

vanzin · 2017-06-07T23:16:21Z

...work-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalShuffleBlockHandler.java

+      mapIdAndReduceIds = new byte[blockIds.length][];
+      if (blockIds.length > 0) {
+        for (int i = 0; i< blockIds.length; i++) {
+          mapIdAndReduceIds[i] = (blockIdParts[2] + "_" + blockIdParts[3]).getBytes();


Instead of storing this as a byte array, how about storing them as ints or longs (depending on what's the actual data type of the id)?

e.g., instead of:

private byte[][] mapIdAndReduceIds;

Which results in blockIds.length + 1 arrays in total, you could have a single one where for each block id you have two entries, one for map id and one for reduce id, or something along those lines.

Yes, I think this is a good idea. In current change, I make it to be int[blockIds.length][2]. I'm not sure if I understand your comment correctly. Please take another look :)

vanzin · 2017-06-07T23:17:16Z

...work-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalShuffleBlockHandler.java

+    private byte[][] mapIdAndReduceIds;
+
+    ManagedBufferIterator(String appId, String execId, String[] blockIds) {
+      this.appId = appId;


Wonder if you see a lot of these in your heap dump too? You could potentially intern appId and execId for some extra memory savings, if you see a lot of those.

@vanzin
There's one appId and execId per stream. I don't see a lot in my heap dump. Do you have any thoughts for interning this? :)

…d and reduceId pairs.

SparkQA · 2017-06-08T06:18:42Z

Test build #77806 has finished for PR 18231 at commit dcf156a.

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

jinxing64 · 2017-06-08T06:32:38Z

@srowen
Thanks a lot looking into this :)
For example: blockId="shuffle_20_1000_2000", it is stored as an String, which costs more than 20 bytes. In this change, it will cost only 8 bytes.

jinxing64 · 2017-06-08T06:38:32Z

@vanzin
Thanks a lot for reviewing this. I refined according to your comments, Please take another look at this when you have time :)

srowen · 2017-06-08T08:24:36Z

That's 12 bytes. Are there millions of these?

jinxing64 · 2017-06-08T09:01:02Z

Actually it's more than 12 bytes.
Yes, there are millions of these. In my heap dump, it's 1.5 G

srowen

Pardon, I'm missing how this saves memory somewhere -- where is a string stored that's now a shorter string?

srowen · 2017-06-08T09:33:35Z

...work-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalShuffleBlockHandler.java

+      }
+      this.shuffleId = blockId0Parts[1];
+      mapIdAndReduceIds = new int[blockIds.length][2];
+      if (blockIds.length > 0) {


This is superfluous

srowen · 2017-06-08T09:33:43Z

...work-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalShuffleBlockHandler.java

+      String[] blockId0Parts = blockIds[0].split("_");
+      if (blockId0Parts.length < 4) {
+        throw new IllegalArgumentException("Unexpected block id format: " + blockIds[0]);
+      } else if (!blockId0Parts[0].equals("shuffle")) {


You don't need the 'else' here

We have some kinds of BlockId, I guess it's better to have a check here and we can parse the blockId correctly.

I think Sean means that since you're throwing in the previous block, else is redundant.

srowen · 2017-06-08T09:34:02Z

...work-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalShuffleBlockHandler.java

    }
  }

+  private class ManagedBufferIterator implements Iterator<ManagedBuffer> {


Why break this out -- it's not necessary for the change right? just for clarity?

I think the iterator is becoming a little bit complicated. So I break this out and give a constructor.

jinxing64 · 2017-06-08T09:46:55Z

@srowen Sorry, I didn't make it clear.

In current code, all blockIds are stored in the iterator. They are released only when the iterator is traversed.
Now I change the String to be two int

srowen · 2017-06-08T09:48:35Z

The current iterator doesn't have any state except for an int. What are you referring to?

jinxing64 · 2017-06-08T09:51:19Z

I mean the blockIds in OpenBlocks, they have reference in iterator.

srowen · 2017-06-08T09:53:19Z

I get it. But that doesn't make the reference in OpenBlocks go away. This only helps anything is msg/msgObj can be garbage collected earlier. Is that the case? right now this is allocating additional memory, not instead of the existing memory.

jinxing64 · 2017-06-08T09:58:26Z

The blockIds cannot be freed because they are referenced in the iterator. In current change they are not. We reference the mapIdAndReduceIds instead. Thus the blockIds in OpenBlocks can be garbage collected.

srowen · 2017-06-08T09:59:56Z

That's not the question though. The question is whether they could be freed even after this change. msg still references it. That's what you need to establish, if only by some empirical testing.

jinxing64 · 2017-06-08T10:05:07Z

there is no where referencing msg, right? I guess the msg will be garbage collected fluently.

srowen · 2017-06-08T10:06:09Z

I'm not clear that's true, no. Not, at least, in the lifetime of the iterator. That's what has to be true for this to help anything. Do you have evidence this is true? for example if you have tests that clearly show the memory is released earlier, that would be good evidence.

jinxing64 · 2017-06-08T10:12:06Z

Yes, I think it's great to do some tests and give a good evidence.

jinxing64 · 2017-06-08T12:23:46Z

@srowen
I did a test to verify this patch.
I wrap a number of blocks inside OpenBlocks and send it to ExternalShuffleBlockHandler.
With this change:
it cost about 133M in the memory; analyzing heap dump, there is only int[][], blockIds is released.
Without this change,:
it cost about 362M in the memory; analyzing heap dump, there is String[].

SparkQA · 2017-06-08T12:45:13Z

Test build #77811 has finished for PR 18231 at commit 1e53262.

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

vanzin · 2017-06-08T16:58:29Z

...work-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalShuffleBlockHandler.java

+    private String execId;
+    private String shuffleId;
+    // An array containing mapId and reduceId pairs.
+    private int[][] mapIdAndReduceIds;


Actually, I mean a single array. e.g.

int[] mapIdAndReduceIds; mapIdAndReduceIds = new int[blockIds.length * 2]; mapIdAndReduceIds[0] = mapId1; mapIdAndReduceIds[1] = reduceId1; mapIdAndReduceIds[2] = mapId2; mapIdAndReduceIds[3] = reduceId2; etc etc etc

Reason being that if you really have millions of these, each "child" array in your two-dimensional array wastes 16 (or 20?) bytes (16 bytes of object overhead + 4 bytes for the array length). Looking in jvisualvm, an empty array actually consumes 24 bytes, so it seems the JVM is aligning things and wasting an extra 4 bytes per array...

SparkQA · 2017-06-09T03:01:50Z

Test build #77831 has finished for PR 18231 at commit 8170c8a.

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

vanzin · 2017-06-09T21:55:00Z

@srowen I don't see any references to the original OpenBlocks message nor to the block id array in the updated code, not sure why do you think there's still a reference somewhere?

srowen · 2017-06-09T22:04:01Z

There isn't a reference here anymore; there could be elsewhere. It sounds like there's good reason to believe there is not another reference hanging around though.

vanzin · 2017-06-09T22:17:52Z

There isn't a reference here anymore; there could be elsewhere.

Only if there was a bug in the RPC layer, since this is an RPC handler and the message should not be referenced by the RPC code after the method returns.

SparkQA · 2017-06-10T02:02:06Z

Test build #77863 has finished for PR 18231 at commit 1e72eab.

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

SparkQA · 2017-06-10T02:37:32Z

Test build #77862 has finished for PR 18231 at commit 5dd0e77.

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

vanzin · 2017-06-13T22:21:19Z

...ork-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalShuffleBlockResolver.java

+   * Obtains a FileSegmentManagedBuffer from (shuffleId, mapId, reduceId). We make assumptions
+   * about how the hash and sort based shuffles store their data.
+   */
+  public ManagedBuffer getBlockData(String appId, String execId, int shuffleId, int mapId,


nit: style. See constructor at top of file for the style when param lists are long.

vanzin · 2017-06-13T22:22:32Z

...ork-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalShuffleBlockResolver.java

-   * assumptions about how the hash and sort based shuffles store their data.
+   * format "shuffle_ShuffleId_MapId_ReduceId" (from ShuffleBlockId).
   */
  public ManagedBuffer getBlockData(String appId, String execId, String blockId) {


Is this method used anywhere else? I only see ExternalShuffleBlockHandler using this class, and it now uses the new method. If only unit tests use this, then remove this method and fix the unit tests.

SparkQA · 2017-06-14T02:36:10Z

Test build #78017 has finished for PR 18231 at commit 3239653.

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

…fix bug.

SparkQA · 2017-06-14T05:55:56Z

Test build #78022 has finished for PR 18231 at commit a2af617.

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

vanzin

One nit otherwise LGTM. I'll leave it overnight in case others want to take a look.

vanzin · 2017-06-14T21:16:30Z

...work-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalShuffleBlockHandler.java

+      }
+      this.shuffleId = Integer.parseInt(blockId0Parts[1]);
+      mapIdAndReduceIds = new int[2 * blockIds.length];
+      for (int i = 0; i< blockIds.length; i++) {


nit: space before <

vanzin · 2017-06-14T21:33:38Z

(also PR title has a typo, should be "redundant")

SparkQA · 2017-06-15T05:10:59Z

Test build #78079 has finished for PR 18231 at commit 6677bc9.

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

jiangxb1987

The change seems reasonable, but the code style still need to be improved. Also cc @cloud-fan to make a pass.

jiangxb1987 · 2017-06-15T14:51:42Z

...work-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalShuffleBlockHandler.java

+      this.appId = appId;
+      this.execId = execId;
+      String[] blockId0Parts = blockIds[0].split("_");
+      if (blockId0Parts.length < 4) {


How about use require(blockId0Parts.length < 4, "Unexpected block id format: " + blockIds[0]) instead?

I was thinking to throw the IllegalArgumentException.
Pardon, I'm not sure how to use require in java.

nvm, didn't notice they are java code.

shall we be more strict and use blockId0Parts.length != 4?

jiangxb1987 · 2017-06-15T14:51:59Z

...work-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalShuffleBlockHandler.java

+      if (blockId0Parts.length < 4) {
+        throw new IllegalArgumentException("Unexpected block id format: " + blockIds[0]);
+      }
+      if (!blockId0Parts[0].equals("shuffle")) {


jiangxb1987 · 2017-06-15T15:00:32Z

...work-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalShuffleBlockHandler.java

+      mapIdAndReduceIds = new int[2 * blockIds.length];
+      for (int i = 0; i < blockIds.length; i++) {
+        String[] blockIdParts = blockIds[i].split("_");
+        if (Integer.parseInt(blockIdParts[1]) != shuffleId) {


blockIdParts[1].toInt != shuffleId ?

It's hard to do that in java, right?

jiangxb1987 · 2017-06-15T15:04:43Z

...work-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalShuffleBlockHandler.java

+      }
+      this.shuffleId = Integer.parseInt(blockId0Parts[1]);
+      mapIdAndReduceIds = new int[2 * blockIds.length];
+      for (int i = 0; i < blockIds.length; i++) {


How about rewrite this to be imperative?

Pardon, could you give an example?

jiangxb1987 · 2017-06-15T15:06:14Z

...ork-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalShuffleBlockResolver.java

      throw new RuntimeException(
        String.format("Executor is not registered (appId=%s, execId=%s)", appId, execId));
    }
-


nit: we should keep the original format.

There are two blank lines originally. I guess it's appropriate to remove one?

jiangxb1987 · 2017-06-15T15:06:42Z

common/network-shuffle/src/test/java/org/apache/spark/network/sasl/SaslIntegrationSuite.java

      };

-      String[] blockIds = { "shuffle_2_3_4", "shuffle_6_7_8" };
+      String[] blockIds = { "shuffle_0_1_2", "shuffle_0_3_4" };


What's the purpose of this change?

With this change ,we cannot shuffle blocks with multiple shuffleIds

jinxing64 · 2017-06-15T15:56:05Z

@jiangxb1987
Thanks a lot for taking time review this pr. More comments are welcome.

cloud-fan · 2017-06-16T04:31:18Z

...work-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalShuffleBlockHandler.java

+      this.shuffleId = Integer.parseInt(blockId0Parts[1]);
+      mapIdAndReduceIds = new int[2 * blockIds.length];
+      for (int i = 0; i < blockIds.length; i++) {
+        String[] blockIdParts = blockIds[i].split("_");


shall we check blockIdParts[0] == "shufle"?

cloud-fan · 2017-06-16T04:33:51Z

...work-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalShuffleBlockHandler.java

+
+    @Override
+    public boolean hasNext() {
+      return index < mapIdAndReduceIds.length / 2;


nit: we can keep a pos, and increase it by 2 in next, so here we can just write pos < mapIdAndReduceIds.length to save a division.

cloud-fan · 2017-06-16T04:36:41Z

LGTM except some minor comments

jinxing64 · 2017-06-16T05:11:00Z

@cloud-fan
Thanks a lot for taking time review this. I refined accordingly :)

SparkQA · 2017-06-16T05:12:41Z

Test build #78155 has started for PR 18231 at commit 2592ef4.

cloud-fan · 2017-06-16T05:33:36Z

...work-shuffle/src/main/java/org/apache/spark/network/shuffle/ExternalShuffleBlockHandler.java

+      this.appId = appId;
+      this.execId = execId;
+      String[] blockId0Parts = blockIds[0].split("_");
+      if (blockId0Parts.length < 4 || !blockId0Parts[0].equals("shuffle")) {


use blockId0Parts.length != 4?

SparkQA · 2017-06-16T05:42:37Z

Test build #78157 has started for PR 18231 at commit 5b0ce67.

jinxing64 · 2017-06-16T07:10:36Z

Jenkins, retest this please

SparkQA · 2017-06-16T09:59:27Z

Test build #78166 has finished for PR 18231 at commit 5b0ce67.

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

cloud-fan · 2017-06-16T12:10:08Z

thanks, merging to master!

jinxing64 · 2017-06-16T12:19:51Z

@cloud–fan
Thanks for merging !

[SPARK-20994] Remove reduant characters in OpenBlocks to save memory …

96d07aa

…for shuffle service.

srowen reviewed Jun 7, 2017

View reviewed changes

vanzin reviewed Jun 7, 2017

View reviewed changes

Fix bug and make mapIdAndReduceIds to be an int array containing mapI…

dcf156a

…d and reduceId pairs.

srowen requested changes Jun 8, 2017

View reviewed changes

refine according to srowen's comments

1e53262

vanzin reviewed Jun 8, 2017

View reviewed changes

make mapIdAndReduceIds a single array.

8170c8a

vanzin reviewed Jun 13, 2017

View reviewed changes

remove getBlockData(String appId, String execId, String blockId) and …

a2af617

…fix bug.

jinxing64 force-pushed the SPARK-20994-v2 branch from 3239653 to a2af617 Compare June 14, 2017 03:11

vanzin reviewed Jun 14, 2017

View reviewed changes

jinxing64 changed the title ~~[SPARK-20994] Remove reduant characters in OpenBlocks to save memory for shuffle service.~~ [SPARK-20994] Remove redundant characters in OpenBlocks to save memory for shuffle service. Jun 15, 2017

fix

6677bc9

jiangxb1987 reviewed Jun 15, 2017

View reviewed changes

cloud-fan reviewed Jun 16, 2017

View reviewed changes

resolve cloud-fan's comments

2592ef4

cloud-fan reviewed Jun 16, 2017

View reviewed changes

blockId0Parts.length != 4

5b0ce67

asfgit closed this in 93dd0c5 Jun 16, 2017

[SPARK-20994] Remove redundant characters in OpenBlocks to save memory for shuffle service. #18231

[SPARK-20994] Remove redundant characters in OpenBlocks to save memory for shuffle service. #18231

Uh oh!

Conversation

jinxing64 commented Jun 7, 2017

What changes were proposed in this pull request?

Uh oh!

jinxing64 commented Jun 7, 2017

Uh oh!

srowen left a comment

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Jun 7, 2017

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 Jun 8, 2017

Uh oh!

jinxing64 commented Jun 8, 2017

Uh oh!

jinxing64 commented Jun 8, 2017

Uh oh!

srowen commented Jun 8, 2017

Uh oh!

jinxing64 commented Jun 8, 2017

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!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

jinxing64 commented Jun 8, 2017

Uh oh!

srowen commented Jun 8, 2017

Uh oh!

jinxing64 commented Jun 8, 2017

Uh oh!

srowen commented Jun 8, 2017

Uh oh!

jinxing64 commented Jun 8, 2017

Uh oh!

srowen commented Jun 8, 2017

Uh oh!

jinxing64 commented Jun 8, 2017

Uh oh!

srowen commented Jun 8, 2017

Uh oh!

jinxing64 commented Jun 8, 2017

Uh oh!

jinxing64 commented Jun 8, 2017

Uh oh!

SparkQA commented Jun 8, 2017

Uh oh!

vanzin Jun 8, 2017 • 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!

SparkQA commented Jun 9, 2017

Uh oh!

vanzin commented Jun 9, 2017

Uh oh!

srowen commented Jun 9, 2017

Uh oh!

vanzin commented Jun 9, 2017

vanzin Jun 8, 2017 •

edited

Loading