Repartition and RepartitionByExpression-spark技术分享

Repartition Logical Operators — Repartition and RepartitionByExpression

Repartition and RepartitionByExpression (repartition operations in short) are unary logical operators that create a new RDD that has exactly numPartitions partitions.

Note	`RepartitionByExpression` is also called distribute operator.

Repartition is the result of coalesce or repartition (with no partition expressions defined) operators.


val rangeAlone = spark.range(5)

scala> rangeAlone.rdd.getNumPartitions
res0: Int = 8

// Repartition the records

val withRepartition = rangeAlone.repartition(numPartitions = 5)

scala> withRepartition.rdd.getNumPartitions
res1: Int = 5

scala> withRepartition.explain(true)
== Parsed Logical Plan ==
Repartition 5, true
+- Range (0, 5, step=1, splits=Some(8))

// ...

== Physical Plan ==
Exchange RoundRobinPartitioning(5)
+- *Range (0, 5, step=1, splits=Some(8))

// Coalesce the records

val withCoalesce = rangeAlone.coalesce(numPartitions = 5)
scala> withCoalesce.explain(true)
== Parsed Logical Plan ==
Repartition 5, false
+- Range (0, 5, step=1, splits=Some(8))

// ...

== Physical Plan ==
Coalesce 5
+- *Range (0, 5, step=1, splits=Some(8))

val rangeAlone = spark.range(5)

scala> rangeAlone.rdd.getNumPartitions

res0: Int = 8

// Repartition the records

val withRepartition = rangeAlone.repartition(numPartitions = 5)

scala> withRepartition.rdd.getNumPartitions

res1: Int = 5

scala> withRepartition.explain(true)

== Parsed Logical Plan ==

Repartition 5, true

+- Range (0, 5, step=1, splits=Some(8))

// ...

== Physical Plan ==

Exchange RoundRobinPartitioning(5)

+- *Range (0, 5, step=1, splits=Some(8))

// Coalesce the records

val withCoalesce = rangeAlone.coalesce(numPartitions = 5)

scala> withCoalesce.explain(true)

== Parsed Logical Plan ==

Repartition 5, false

+- Range (0, 5, step=1, splits=Some(8))

// ...

== Physical Plan ==

Coalesce 5

+- *Range (0, 5, step=1, splits=Some(8))

RepartitionByExpression is the result of the following operators:

Dataset.repartition operator (with explicit partition expressions defined)
Dataset.repartitionByRange
DISTRIBUTE BY SQL clause.


// RepartitionByExpression
// 1) Column-based partition expression only
scala> rangeAlone.repartition(partitionExprs = 'id % 2).explain(true)
== Parsed Logical Plan ==
'RepartitionByExpression [('id % 2)], 200
+- Range (0, 5, step=1, splits=Some(8))

// ...

== Physical Plan ==
Exchange hashpartitioning((id#10L % 2), 200)
+- *Range (0, 5, step=1, splits=Some(8))

// 2) Explicit number of partitions and partition expression
scala> rangeAlone.repartition(numPartitions = 2, partitionExprs = 'id % 2).explain(true)
== Parsed Logical Plan ==
'RepartitionByExpression [('id % 2)], 2
+- Range (0, 5, step=1, splits=Some(8))

// ...

== Physical Plan ==
Exchange hashpartitioning((id#10L % 2), 2)
+- *Range (0, 5, step=1, splits=Some(8))

// RepartitionByExpression

// 1) Column-based partition expression only

scala> rangeAlone.repartition(partitionExprs = 'id % 2).explain(true)

== Parsed Logical Plan ==

'RepartitionByExpression [('id % 2)], 200

+- Range (0, 5, step=1, splits=Some(8))

// ...

== Physical Plan ==

Exchange hashpartitioning((id#10L % 2), 200)

+- *Range (0, 5, step=1, splits=Some(8))

// 2) Explicit number of partitions and partition expression

scala> rangeAlone.repartition(numPartitions = 2, partitionExprs = 'id % 2).explain(true)

== Parsed Logical Plan ==

'RepartitionByExpression [('id % 2)], 2

+- Range (0, 5, step=1, splits=Some(8))

// ...

== Physical Plan ==

Exchange hashpartitioning((id#10L % 2), 2)

+- *Range (0, 5, step=1, splits=Some(8))

Repartition and RepartitionByExpression logical operators are described by:

shuffle flag
target number of partitions

Note	BasicOperators strategy resolves `Repartition` to ShuffleExchangeExec (with RoundRobinPartitioning partitioning scheme) or CoalesceExec physical operators per shuffle — enabled or not, respectively.

Note	BasicOperators strategy resolves `RepartitionByExpression` to ShuffleExchangeExec physical operator with HashPartitioning partitioning scheme.

Repartition Operation Optimizations

CollapseRepartition logical optimization collapses adjacent repartition operations.
Repartition operations allow FoldablePropagation and PushDownPredicate logical optimizations to “push through”.
PropagateEmptyRelation logical optimization may result in an empty LocalRelation for repartition operations.

Repartition and RepartitionByExpression

Repartition Logical Operators — Repartition and RepartitionByExpression

Repartition Operation Optimizations

相关推荐

欢迎关注：spark技术分享

热门标签

近期文章

分类目录

关注公众号：spark技术分享

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

微信扫一扫打赏

QQ咨询

回顶部