[SPARK-19426][SQL] Custom coalescer for Dataset #18861

maropu · 2017-08-06T16:16:55Z

What changes were proposed in this pull request?

This pr added a new API for coalesce in Dataset; users can specify the custom coalescer which reduces an input Dataset into fewer partitions. This coalescer implementation is the same with the one in RDD#coalesce added in #11865 (SPARK-14042).

This is the rework of #16766.

How was this patch tested?

Added tests in DatasetSuite.

SparkQA · 2017-08-06T16:34:23Z

Test build #80306 has finished for PR 18861 at commit bb7f9b0.

This patch fails to generate documentation.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class PartitionCoalesce(numPartitions: Int, coalescer: PartitionCoalescer, child: LogicalPlan)
case class CoalesceExec(numPartitions: Int, child: SparkPlan, coalescer: Option[PartitionCoalescer])

SparkQA · 2017-08-06T19:50:28Z

Test build #80307 has finished for PR 18861 at commit 253bdcc.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class PartitionCoalesce(numPartitions: Int, coalescer: PartitionCoalescer, child: LogicalPlan)
case class CoalesceExec(numPartitions: Int, child: SparkPlan, coalescer: Option[PartitionCoalescer])

maropu · 2017-08-07T00:57:15Z

@gatorsmile ok

gatorsmile · 2017-08-07T04:22:39Z

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

@@ -571,7 +570,8 @@ case class UnionExec(children: Seq[SparkPlan]) extends SparkPlan {
 * current upstream partitions will be executed in parallel (per whatever
 * the current partitioning is).
 */
-case class CoalesceExec(numPartitions: Int, child: SparkPlan) extends UnaryExecNode {
+case class CoalesceExec(numPartitions: Int, child: SparkPlan, coalescer: Option[PartitionCoalescer])


Could you add the parm description of coalescer? also update function descriptions? Thanks~!

gatorsmile · 2017-08-07T04:29:06Z

core/src/test/scala/org/apache/spark/rdd/RDDSuite.scala

          addPartition(partition, splitSize)
          index += 1
        } else {
-          updateGroups
+          updateGroups()


All the above changes are not related to this PR, right?

Yea, I just left the changes of the original author (probably refactoring stuffs?) ..., better remove this?

I am fine about this, but it might confuse the others. Maybe just remove them in this PR? You can submit a separate PR later.

ok, I'll drop these from this pr.

gatorsmile · 2017-08-07T04:29:27Z

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

+ * supplying a `PartitionCoalescer` to control the behavior of the partitioning.
+ */
+case class PartitionCoalesce(numPartitions: Int, coalescer: PartitionCoalescer, child: LogicalPlan)
+  extends UnaryNode {


Adding new logical nodes also needs the updates in multiple different components. (e.g., Optimizer).

Is that possible to reuse the existing node Repartition?

yea, I think so. I'll try and plz give me days to do so.

SparkQA · 2017-08-07T07:04:51Z

Test build #80321 has finished for PR 18861 at commit c0306d3.

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

SparkQA · 2017-08-07T07:04:51Z

Test build #80320 has finished for PR 18861 at commit 413b0eb.

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

SparkQA · 2017-08-07T07:28:55Z

Test build #80327 has finished for PR 18861 at commit 83ac85f.

This patch fails Scala style tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class Repartition(

SparkQA · 2017-08-07T10:44:51Z

Test build #80328 has finished for PR 18861 at commit 3b4c679.

This patch passes all tests.
This patch merges cleanly.
This patch adds the following public classes (experimental):
case class Repartition(

maropu · 2017-08-07T10:47:55Z

@gatorsmile ok, could you check again?

gatorsmile · 2017-08-07T21:20:56Z

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

+    numPartitions: Int,
+    shuffle: Boolean,
+    child: LogicalPlan,
+    coalescer: Option[PartitionCoalescer] = None)
  extends RepartitionOperation {
  require(numPartitions > 0, s"Number of partitions ($numPartitions) must be positive.")


Add a new require here?

require(!shuffle || coalescer.isEmpty, "Custom coalescer is not allowed for repartition(shuffle=true)")

gatorsmile · 2017-08-07T21:21:52Z

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

+   * @since 2.3.0
+   */
+  def coalesce(numPartitions: Int, userDefinedCoalescer: Option[PartitionCoalescer])
+    : Dataset[T] = withTypedPlan {


def coalesce( numPartitions: Int, userDefinedCoalescer: Option[PartitionCoalescer]): Dataset[T] = withTypedPlan {

gatorsmile · 2017-08-07T21:31:35Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala

-      case (false, true) => if (r.numPartitions >= child.numPartitions) child else r
-      case _ => r.copy(child = child.child)
-    }
+    case r @ Repartition(_, _, child: RepartitionOperation, None) =>


sorry, my bad. Fixed.

SparkQA · 2017-08-08T05:49:38Z

Test build #80373 has finished for PR 18861 at commit d7392c4.

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

maropu · 2017-08-08T05:50:45Z

@gatorsmile ok, fixed.

maropu · 2017-08-10T14:03:35Z

@gatorsmile ping

gatorsmile · 2017-08-14T06:38:48Z

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala

@@ -596,14 +596,17 @@ object CollapseProject extends Rule[LogicalPlan] {
 object CollapseRepartition extends Rule[LogicalPlan] {


Please also add new test cases to CollapseRepartitionSuite for the changes in this rule.

jiangxb1987 · 2017-10-03T14:42:57Z

@maropu any update on this issue?

maropu · 2017-10-03T14:51:28Z

@gatorsmile better to close this pr and the jira?

gatorsmile · 2017-10-03T23:56:56Z

Yeah, maybe close it first. Thanks!

maropu · 2017-10-04T00:18:39Z

ok, thanks!

Implement custom coalesce

9abcd75

Add more fixes

253bdcc

maropu force-pushed the pr16766 branch from bb7f9b0 to 253bdcc Compare August 6, 2017 16:44

gatorsmile reviewed Aug 7, 2017

View reviewed changes

Fix minor fixes

c0306d3

maropu force-pushed the pr16766 branch from 413b0eb to c0306d3 Compare August 7, 2017 06:03

Drop PartitionCoalesce and use Reparititon

3b4c679

maropu force-pushed the pr16766 branch from 83ac85f to 3b4c679 Compare August 7, 2017 07:35

gatorsmile reviewed Aug 7, 2017

View reviewed changes

Apply comments

d7392c4

gatorsmile reviewed Aug 14, 2017

View reviewed changes

maropu closed this Oct 4, 2017

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[SPARK-19426][SQL] Custom coalescer for Dataset #18861

[SPARK-19426][SQL] Custom coalescer for Dataset #18861

maropu commented Aug 6, 2017

SparkQA commented Aug 6, 2017

SparkQA commented Aug 6, 2017

maropu commented Aug 7, 2017

gatorsmile Aug 7, 2017

maropu Aug 7, 2017

gatorsmile Aug 7, 2017

maropu Aug 7, 2017

gatorsmile Aug 7, 2017

maropu Aug 7, 2017

gatorsmile Aug 7, 2017

maropu Aug 7, 2017

SparkQA commented Aug 7, 2017

SparkQA commented Aug 7, 2017

SparkQA commented Aug 7, 2017

SparkQA commented Aug 7, 2017

maropu commented Aug 7, 2017

gatorsmile Aug 7, 2017

maropu Aug 8, 2017

gatorsmile Aug 7, 2017

gatorsmile Aug 7, 2017

maropu Aug 8, 2017

SparkQA commented Aug 8, 2017

maropu commented Aug 8, 2017

maropu commented Aug 10, 2017

gatorsmile Aug 14, 2017

maropu Aug 14, 2017

jiangxb1987 commented Oct 3, 2017

maropu commented Oct 3, 2017

gatorsmile commented Oct 3, 2017

maropu commented Oct 4, 2017

		@@ -596,14 +596,17 @@ object CollapseProject extends Rule[LogicalPlan] {
		object CollapseRepartition extends Rule[LogicalPlan] {

[SPARK-19426][SQL] Custom coalescer for Dataset #18861

[SPARK-19426][SQL] Custom coalescer for Dataset #18861

Conversation

maropu commented Aug 6, 2017

What changes were proposed in this pull request?

How was this patch tested?

SparkQA commented Aug 6, 2017

SparkQA commented Aug 6, 2017

maropu commented Aug 7, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Aug 7, 2017

SparkQA commented Aug 7, 2017

SparkQA commented Aug 7, 2017

SparkQA commented Aug 7, 2017

maropu commented Aug 7, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

SparkQA commented Aug 8, 2017

maropu commented Aug 8, 2017

maropu commented Aug 10, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jiangxb1987 commented Oct 3, 2017

maropu commented Oct 3, 2017

gatorsmile commented Oct 3, 2017

maropu commented Oct 4, 2017