[SPARK-21782][Core] Repartition creates skews when numPartitions is a power of 2

[megaserg]

Problem

When an RDD (particularly with a low item-per-partition ratio) is repartitioned to numPartitions = power of 2, the resulting partitions are very uneven-sized, due to using fixed seed to initialize PRNG, and using the PRNG only once. See details in https://issues.apache.org/jira/browse/SPARK-21782

What changes were proposed in this pull request?

Instead of directly using 0, 1, 2,... seeds to initialize Random, hash them with scala.util.hashing.byteswap32().

How was this patch tested?

build/mvn -Dtest=none -DwildcardSuites=org.apache.spark.rdd.RDDSuite test

[srowen]

You definitely can't do this. Evaluating the RDD becomes nondeterministic

[megaserg]

Sorry, I edited the pull request body. The @srowen's comment above was referring to the initial version, where I proposed using default, non-deterministic constructor for Random().

[SparkQA]

Test build #3891 has finished for PR 18990 at commit bee7fca.

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

[srowen]

I'll leave it open a bit for more comments, but in theory this change is fine as nobody should depend on the exact output. In practice it might change the exact output of a shuffle stage. But no tests failed, which is evidence that it has very little if any practical impact.

[jiangxb1987]

LGTM, cc @yanboliang @cloud-fan

[viirya]

LGTM. I agree that in theory there is no reason we should depend on the exact shuffle distribution here. It should be beneficial to have a more even distribution.

[atronchi]

This issue appears to remain at large for the dataframe API which is used more broadly than RDD. What would it take to extend the fix to the dataframe API?

I verified this on Spark 3.2 using df.repartition(1024) on a dataframe with ~200k rows, which resulted in almost 30% EMPTY partitions, and the below shown skew of the remaining non-empty ones.

[srowen]

@atronchi what is "df" here? I couldn't reproduce that with a DF of 200K simple rows

[cloud-fan]

It should have been fixed in 3.2+: #37855

[atronchi]

df is a cached dataframe containing ~200k rows. @srowen what Spark version did you test with?

[srowen]

I was using 3.3.2