[SPARK-22614] Dataset API: repartitionByRange(...) #19828

adrian-ionescu · 2017-11-27T15:52:19Z

What changes were proposed in this pull request?

This PR introduces a way to explicitly range-partition a Dataset. So far, only round-robin and hash partitioning were possible via df.repartition(...), but sometimes range partitioning might be desirable: e.g. when writing to disk, for better compression without the cost of global sort.

The current implementation piggybacks on the existing RepartitionByExpression LogicalPlan and simply adds the following logic: If its expressions are of type SortOrder, then it will do RangePartitioning; otherwise HashPartitioning. This was by far the least intrusive solution I could come up with.

How was this patch tested?

Unit test for RepartitionByExpression changes, a test to ensure we're not changing the behavior of existing .repartition() and a few end-to-end tests in DataFrameSuite.

hvanhovell · 2017-11-27T16:07:39Z

ok to test

hvanhovell · 2017-11-27T16:09:51Z

sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala

   */
  @scala.annotation.varargs
  def repartition(numPartitions: Int, partitionExprs: Column*): Dataset[T] = withTypedPlan {
+    partitionExprs.find(_.expr.isInstanceOf[SortOrder]).foreach { sortOrder =>


Use collect or filter? That way we can show all offending columns.

hvanhovell · 2017-11-27T16:10:50Z

...alyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala

+  val (sortOrder, nonSortOrder) = partitionExpressions.partition(_.isInstanceOf[SortOrder])
+
+  require(sortOrder.isEmpty || nonSortOrder.isEmpty,
+    s"""${getClass.getSimpleName} expects that either all its `partitionExpressions` are of type


Do you want this to be a multiline message? it makes sense to put the sort order and non sort order on new lines.

hvanhovell · 2017-11-27T16:12:26Z

sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala

+        // RepartitionByExpression's constructor verifies that either all expressions are
+        // of type SortOrder, in which case we're doing RangePartitioning, or none of them are,
+        // in which case we're doing HashPartitioning.
+        val partitioning = if (expressions.forall(_.isInstanceOf[SortOrder])) {


We have discussed this before, but to me it makes slightly more sense to add this logic to the RepartitionByExpression plan.

Oh, and it also makes it easier to unit test this code :)

hvanhovell · 2017-11-27T16:14:51Z

sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala

+
+    // .repartitionByRange() assumes .asc by default if no explicit sort order is specified
+    checkAnswer(
+      data2d.toDF("a", "b").repartitionByRange(data1d.size, $"a".desc, $"b")


data1d.size?

hvanhovell · 2017-11-27T16:15:44Z

add to whitelist

hvanhovell · 2017-11-27T16:24:39Z

@adrian-ionescu this looks pretty good. I left a few small comments.

gatorsmile · 2017-11-27T17:38:41Z

sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala

    }
  }
+
+  test("repartitionByRange") {


Move it to DataFrameSuite?

gatorsmile · 2017-11-27T17:53:38Z

sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala

+   * @since 2.3.0
+   */
+  @scala.annotation.varargs
+  def repartitionByRange(numPartitions: Int, partitionExprs: Column*): Dataset[T] = withTypedPlan {


Open a JIRA for adding the corresponding API in PySpark?

Good call! Raised SPARK-22624.

gatorsmile · 2017-11-27T17:59:18Z

sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala

+      col.expr match {
+        case expr: SortOrder =>
+          expr
+        case expr: Expression =>


What happened if we have a SortOrder that is not in the root node of expr?

data1d.toDF("val").repartitionByRange(data1d.size, $"val".desc + 1) .select(spark_partition_id().as("id"), $"val").show()

org.apache.spark.sql.catalyst.errors.package$TreeNodeException: execute, tree: Exchange rangepartitioning((val#236 DESC NULLS LAST + 1) ASC NULLS FIRST, 10) +- LocalTableScan [val#236] at org.apache.spark.sql.catalyst.errors.package$.attachTree(package.scala:56) at org.apache.spark.sql.execution.exchange.ShuffleExchangeExec.doExecute(ShuffleExchangeExec.scala:116) at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:117) at org.apache.spark.sql.execution.SparkPlan$$anonfun$execute$1.apply(SparkPlan.scala:113)

This is a more generic problem right? I think a similar error gets thrown if you do something like this: spark.range(10).select($"id".asc + 1).show()

Let's fix that in a different ticket.

Cannot evaluate expression: input[0, bigint, false] ASC NULLS FIRST java.lang.UnsupportedOperationException: Cannot evaluate expression: input[0, bigint, false] ASC NULLS FIRST at org.apache.spark.sql.catalyst.expressions.Unevaluable$class.doGenCode(Expression.scala:259) at org.apache.spark.sql.catalyst.expressions.SortOrder.doGenCode(SortOrder.scala:60)

Yeah. It also does not work.

The error is slightly different because the project is whole stage code generated.

gatorsmile · 2017-11-27T18:02:32Z

...alyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala

+  require(partitionExpressions.nonEmpty, "At least one partition-by expression must be specified.")
+
+  val partitioning: Partitioning = {
+    val (sortOrder, nonSortOrder) = partitionExpressions.partition(_.isInstanceOf[SortOrder])


Still the same question. What happened when the SortOrder is not at the root node.

It's going to follow the HashPartitioning path and eventually lead to a "Cannot evaluate expression" exception, just like it would presently do if you tried running df.repartition($"col".asc + 1) or df.sort($"col".asc + 1)

SparkQA · 2017-11-27T19:02:40Z

Test build #84223 has finished for PR 19828 at commit 950b3dc.

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

SparkQA · 2017-11-27T19:07:42Z

Test build #84222 has finished for PR 19828 at commit 950b3dc.

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

SparkQA · 2017-11-27T20:26:05Z

Test build #84225 has finished for PR 19828 at commit 671e9e4.

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

gatorsmile · 2017-11-28T07:33:38Z

LGTM.

Like what @hvanhovell suggested, we can fix it in the follow-up PR. Thanks!

SparkQA · 2017-11-28T08:05:01Z

Test build #84247 has finished for PR 19828 at commit fe690fc.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

adrian-ionescu · 2017-11-28T08:40:09Z

[error] running /home/jenkins/workspace/SparkPullRequestBuilder/build/sbt -Phadoop-2.6 -Phive-thriftserver -Phive -Dtest.exclude.tags=org.apache.spark.tags.ExtendedHiveTest,org.apache.spark.tags.ExtendedYarnTest hive-thriftserver/test mllib/test hive/test repl/test examples/test sql/test sql-kafka-0-10/test catalyst/test ; process was terminated by signal 9

Does not seem to be related.. can we retrigger the tests?

adrian-ionescu · 2017-11-28T08:45:22Z

jenkins retest this please

SparkQA · 2017-11-28T11:31:54Z

Test build #84252 has finished for PR 19828 at commit fe690fc.

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

SparkQA · 2017-11-28T15:39:31Z

Test build #84258 has finished for PR 19828 at commit 66b192d.

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

gatorsmile · 2017-11-28T19:58:59Z

...alyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala

+      s"${getClass.getSimpleName} expects that either all its `partitionExpressions` are of type " +
+        "`SortOrder`, which means `RangePartitioning`, or none of them are `SortOrder`, which " +
+        "means `HashPartitioning`. In this case we have:" +
+      s""""


nice catch :)

This still exists after we revert the previous changes.

gatorsmile · 2017-11-28T20:53:59Z

sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala

+
+  /**
+   * Returns a new Dataset partitioned by the given partitioning expressions into
+   * `numPartitions`. The resulting Dataset is range partitioned.


Could you update this to describe the latest change?

SparkQA · 2017-11-29T15:40:06Z

Test build #84296 has finished for PR 19828 at commit 012d617.

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

gatorsmile · 2017-11-29T17:25:44Z

sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala

+  }
+
+  /**
+   * Returns a new Dataset that is hash partitioned by the given expressions into `numPartitions`.


hash -> range

SparkQA · 2017-11-29T20:18:25Z

Test build #84308 has finished for PR 19828 at commit 60ec0e3.

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

This reverts commit 60ec0e3

This reverts commit 012d617.

This reverts commit 66b192d

…mpty expr list

SparkQA · 2017-11-30T16:38:25Z

Test build #84338 has finished for PR 19828 at commit 67ca139.

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

gatorsmile · 2017-11-30T17:40:39Z

...alyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala

+      s"${getClass.getSimpleName} expects that either all its `partitionExpressions` are of type " +
+        "`SortOrder`, which means `RangePartitioning`, or none of them are `SortOrder`, which " +
+        "means `HashPartitioning`. In this case we have:" +
+      s""""


This still exists after we revert the previous changes.

gatorsmile · 2017-11-30T18:18:12Z

...alyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala


  require(numPartitions > 0, s"Number of partitions ($numPartitions) must be positive.")

-  require(partitionExpressions.nonEmpty, "At least one partition-by expression must be specified.")


Just for safety, also keep this change?

That would change the current behavior of .repartition(numPartitions, Seq.empty: _*) and I'd like to avoid that.

In fact, I've just raised a separate ticket about the latter: SPARK-22665

SparkQA · 2017-11-30T21:19:23Z

Test build #84347 has finished for PR 19828 at commit 012baa0.

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

gatorsmile · 2017-11-30T23:41:09Z

LGTM

Thanks! Merged to master.

adrian-ionescu added 2 commits November 27, 2017 16:22

repartitionByRange() + repartition() with SortOrders

08527b0

avoid changing semantics for .repartition()

950b3dc

hvanhovell reviewed Nov 27, 2017

View reviewed changes

review feedback: val partition = ... + unit test

671e9e4

gatorsmile reviewed Nov 27, 2017

View reviewed changes

move test to DataFrameSuite + minor ws

fe690fc

handle empty partition-by expr list as RoundRobin

66b192d

gatorsmile reviewed Nov 28, 2017

View reviewed changes

update docs

012d617

gatorsmile reviewed Nov 29, 2017

View reviewed changes

fix typo: hash -> range

60ec0e3

Revert "fix typo: hash -> range"

e0bda1d

This reverts commit 60ec0e3

adrian-ionescu added 3 commits November 30, 2017 12:24

Revert "update docs"

2f0f8f4

This reverts commit 012d617.

Revert "handle empty partition-by expr list as RoundRobin"

f6cd388

This reverts commit 66b192d

avoid changing repartition() and make repartitionByRange() throw on e…

67ca139

…mpty expr list

gatorsmile approved these changes Nov 30, 2017

View reviewed changes

fix triple quotes

012baa0

asfgit closed this in f5f8e84 Nov 30, 2017


		require(numPartitions > 0, s"Number of partitions ($numPartitions) must be positive.")

		require(partitionExpressions.nonEmpty, "At least one partition-by expression must be specified.")

[SPARK-22614] Dataset API: repartitionByRange(...) #19828

[SPARK-22614] Dataset API: repartitionByRange(...) #19828

Uh oh!

Conversation

adrian-ionescu commented Nov 27, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

hvanhovell commented Nov 27, 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!

Choose a reason for hiding this comment

Uh oh!

hvanhovell commented Nov 27, 2017

Uh oh!

hvanhovell commented Nov 27, 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!

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!

adrian-ionescu Nov 28, 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 Nov 27, 2017

Uh oh!

SparkQA commented Nov 27, 2017

Uh oh!

SparkQA commented Nov 27, 2017

Uh oh!

gatorsmile commented Nov 28, 2017

Uh oh!

SparkQA commented Nov 28, 2017

Uh oh!

adrian-ionescu commented Nov 28, 2017

Uh oh!

adrian-ionescu commented Nov 28, 2017

Uh oh!

SparkQA commented Nov 28, 2017

Uh oh!

SparkQA commented Nov 28, 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 Nov 29, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Nov 29, 2017

Uh oh!

SparkQA commented Nov 30, 2017

Uh oh!

Choose a reason for hiding this comment

adrian-ionescu commented Nov 27, 2017 •

edited

Loading

adrian-ionescu Nov 28, 2017 •

edited

Loading

adrian-ionescu Nov 30, 2017 •

edited

Loading