[SPARK-9189][CORE] Takes locality and the sum of partition length into account when partition is instance of HadoopPartition in operator coalesce

[watermen]

Before:
Takes locality and the number of partitions into account in operator coalesce.

After:
Takes locality and the sum of partition length(part1.len + part2.len + ... + partN.len) into account when partition is instance of HadoopPartition in operator coalesce.

To make the data size of partition more balanced.

/cc @liancheng @scwf

[scwf]

@watermen i think this should be [SPARK-9189][CORE]

[andrewor14]

add to whitelist

[SparkQA]

Test build #37909 has finished for PR 7536 at commit cb72d0f.

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

[watermen]

@andrewor14 retest it?
[error] SERVER ERROR: Service Temporarily Unavailable url=http://maven.twttr.com/org/apache/hadoop/hadoop-yarn-server/2.2.0/hadoop-yarn-server-2.2.0.jar

[liancheng]

@watermen You can retest it yourself now as Andrew has put you into the whitelist.

[liancheng]

Jenkins is being shut down for maintenance purposes. You may retest this PR after it comes back.

[watermen]

retest this please

[SparkQA]

Test build #42 has finished for PR 7536 at commit cb72d0f.

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

[SparkQA]

Test build #37951 has finished for PR 7536 at commit cb72d0f.

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

[watermen]

retest this please

[SparkQA]

Test build #38004 has finished for PR 7536 at commit cb72d0f.

• This patch fails some tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #47 has finished for PR 7536 at commit cb72d0f.

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

[scwf]

retest this please

[SparkQA]

Test build #57 has finished for PR 7536 at commit cb72d0f.

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

[SparkQA]

Test build #38066 has finished for PR 7536 at commit cb72d0f.

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

[watermen]

retest this please

[SparkQA]

Test build #38141 has finished for PR 7536 at commit cb72d0f.

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

[SparkQA]

Test build #67 has finished for PR 7536 at commit cb72d0f.

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

[watermen]

retest this please

[SparkQA]

Test build #108 has finished for PR 7536 at commit cb72d0f.

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

[SparkQA]

Test build #38473 has finished for PR 7536 at commit cb72d0f.

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

[hellertime]

Shouldn't this be val groupLen2 = groupArr(r2)? Otherwise groupLen1 will aways equal groupLen2.

[watermen]

@hellertime Yes, I'll fix this bug, thanks.

[watermen]

retest this please

[SparkQA]

Test build #244 has finished for PR 7536 at commit 5623370.

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

[SparkQA]

Test build #39997 has finished for PR 7536 at commit 5623370.

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

[watermen]

@andrewor14 @srowen Any more comment on this?

[srowen]

That seems reasonable to me, although I am not sure if inputLength is necessarily set in all cases? this is just because I don't know the code. Also are there other places that need a similar treatment?

[andrewor14]

@watermen can you add a unit test for this? The high level motivation sounds reasonable to me, but like @srowen I'm not familiar enough with the Hadoop code to merge this. Perhaps @tgravescs would have a better idea?

[tgravescs]

On the hadoop size getLength() is part of the required interface so the function would be there. With all the inputs formats I've seen this is always set to something reasonable but anyone can write a custom input format. I could see cases where someone has an input format that they don't know the size or its more expensive to compute the size then just to fetch it. You could simply add a check for 0 and fall back to the # of partitions.

@watermen have you run this on real cluster with skewed data to see if it makes a difference? what input formats have you used?

if there are thousands (or tens of thousands) of partitions and you are coalescing into small # of buckets we are now potentially calculating the length in every group over and over again. did you test to see how long that takes vs just checking the size of the array? I'm guessing that isn't to bad but it doesn't hurt to verify.

[srowen]

(PS Yeah I meant that I wasn't sure whether getLength always returned a meaningful value, as I sort of distantly remember a problem with this, but, it might be imagining it.)