spark-sql-spark技术分享-第14页

GetCurrentDatabase Logical Optimization

GetCurrentDatabase is a base logical optimization that gives the current database for current_database SQL function.

GetCurrentDatabase is part of the Finish Analysis once-executed batch in the standard batches of the Catalyst Optimizer.

GetCurrentDatabase is simply a Catalyst rule for transforming logical plans, i.e. Rule[LogicalPlan].



val q = sql("SELECT current_database() AS db")
val analyzedPlan = q.queryExecution.analyzed

scala> println(analyzedPlan.numberedTreeString)
00 Project [current_database() AS db#22]
01 +- OneRowRelation

import org.apache.spark.sql.catalyst.optimizer.GetCurrentDatabase

val afterGetCurrentDatabase = GetCurrentDatabase(spark.sessionState.catalog)(analyzedPlan)
scala> println(afterGetCurrentDatabase.numberedTreeString)
00 Project [default AS db#22]
01 +- OneRowRelation

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

val q = sql("SELECT current_database() AS db")

val analyzedPlan = q.queryExecution.analyzed

scala> println(analyzedPlan.numberedTreeString)

00 Project [current_database() AS db#22]

01 +- OneRowRelation

import org.apache.spark.sql.catalyst.optimizer.GetCurrentDatabase

val afterGetCurrentDatabase = GetCurrentDatabase(spark.sessionState.catalog)(analyzedPlan)

scala> println(afterGetCurrentDatabase.numberedTreeString)

00 Project [default AS db#22]

01 +- OneRowRelation

Note

GetCurrentDatabase corresponds to SQL’s current_database() function.

You can access the current database in Scala using



scala> val database = spark.catalog.currentDatabase
database: String = default

1

2

3

4

5

6

scala> val database = spark.catalog.currentDatabase

database: String = default

Executing Rule — `apply` Method



apply(plan: LogicalPlan): LogicalPlan

1

2

3

4

5

apply(plan: LogicalPlan): LogicalPlan

Note	`apply` is part of the Rule Contract to execute (apply) a rule on a TreeNode (e.g. LogicalPlan).

apply…FIXME

EliminateView Logical Optimization

EliminateView is a base logical optimization that removes (eliminates) View logical operators from a logical query plan.

EliminateView is part of the Finish Analysis once-executed batch in the standard batches of the Catalyst Optimizer.

EliminateView is simply a Catalyst rule for transforming logical plans, i.e. Rule[LogicalPlan].



val name = "demo_view"
sql(s"CREATE OR REPLACE VIEW $name COMMENT 'demo view' AS VALUES 1,2")
assert(spark.catalog.tableExists(name))

val q = spark.table(name)

val analyzedPlan = q.queryExecution.analyzed
scala> println(analyzedPlan.numberedTreeString)
00 SubqueryAlias demo_view
01 +- View (`default`.`demo_view`, [col1#37])
02    +- Project [cast(col1#38 as int) AS col1#37]
03       +- LocalRelation [col1#38]

import org.apache.spark.sql.catalyst.analysis.EliminateView
val afterEliminateView = EliminateView(analyzedPlan)
// Notice no View operator
scala> println(afterEliminateView.numberedTreeString)
00 SubqueryAlias demo_view
01 +- Project [cast(col1#38 as int) AS col1#37]
02    +- LocalRelation [col1#38]

// TIP: You may also want to use EliminateSubqueryAliases to eliminate SubqueryAliases

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

val name = "demo_view"

sql(s"CREATE OR REPLACE VIEW $name COMMENT 'demo view' AS VALUES 1,2")

assert(spark.catalog.tableExists(name))

val q = spark.table(name)

val analyzedPlan = q.queryExecution.analyzed

scala> println(analyzedPlan.numberedTreeString)

00 SubqueryAlias demo_view

01 +- View (`default`.`demo_view`, [col1#37])

02 +- Project [cast(col1#38 as int) AS col1#37]

03 +- LocalRelation [col1#38]

import org.apache.spark.sql.catalyst.analysis.EliminateView

val afterEliminateView = EliminateView(analyzedPlan)

// Notice no View operator

scala> println(afterEliminateView.numberedTreeString)

00 SubqueryAlias demo_view

01 +- Project [cast(col1#38 as int) AS col1#37]

02 +- LocalRelation [col1#38]

// TIP: You may also want to use EliminateSubqueryAliases to eliminate SubqueryAliases

Executing Rule — `apply` Method



apply(plan: LogicalPlan): LogicalPlan

1

2

3

4

5

apply(plan: LogicalPlan): LogicalPlan

Note	`apply` is part of the Rule Contract to execute (apply) a rule on a TreeNode (e.g. LogicalPlan).

apply simply removes (eliminates) View unary logical operators from the input logical plan and replaces them with their child logical operator.

apply throws an AssertionError when the output schema of the View operator does not match the output schema of the child logical operator.



assertion failed: The output of the child [output] is different from the view output [output]

1

2

3

4

5

assertion failed: The output of the child [output] is different from the view output [output]

Note	The assertion should not really happen since AliasViewChild logical analysis rule is executed earlier and takes care of not allowing for such difference in the output schema (by throwing an `AnalysisException` earlier).

EliminateSubqueryAliases Logical Optimization

EliminateSubqueryAliases is a base logical optimization that removes (eliminates) SubqueryAlias logical operators from a logical query plan.

EliminateSubqueryAliases is part of the Finish Analysis once-executed batch in the standard batches of the Catalyst Optimizer.

EliminateSubqueryAliases is simply a Catalyst rule for transforming logical plans, i.e. Rule[LogicalPlan].



// Using Catalyst DSL
import org.apache.spark.sql.catalyst.dsl.plans._
val t1 = table("t1")
val logicalPlan = t1.subquery('a)

import org.apache.spark.sql.catalyst.analysis.EliminateSubqueryAliases
val afterEliminateSubqueryAliases = EliminateSubqueryAliases(logicalPlan)
scala> println(afterEliminateSubqueryAliases.numberedTreeString)
00 'UnresolvedRelation `t1`

1

2

3

4

5

6

7

8

9

10

11

12

13

// Using Catalyst DSL

import org.apache.spark.sql.catalyst.dsl.plans._

val t1 = table("t1")

val logicalPlan = t1.subquery('a)

import org.apache.spark.sql.catalyst.analysis.EliminateSubqueryAliases

val afterEliminateSubqueryAliases = EliminateSubqueryAliases(logicalPlan)

scala> println(afterEliminateSubqueryAliases.numberedTreeString)

00 'UnresolvedRelation `t1`

Executing Rule — `apply` Method



apply(plan: LogicalPlan): LogicalPlan

1

2

3

4

5

apply(plan: LogicalPlan): LogicalPlan

Note	`apply` is part of the Rule Contract to execute (apply) a rule on a TreeNode (e.g. LogicalPlan).

apply simply removes (eliminates) SubqueryAlias unary logical operators from the input logical plan.

EliminateSerialization Logical Optimization

EliminateSerialization is a base logical optimization that optimizes logical plans with DeserializeToObject (after SerializeFromObject or TypedFilter), AppendColumns (after SerializeFromObject), TypedFilter (after SerializeFromObject) logical operators.

EliminateSerialization is part of the Operator Optimization before Inferring Filters fixed-point batch in the standard batches of the Catalyst Optimizer.

EliminateSerialization is simply a Catalyst rule for transforming logical plans, i.e. Rule[LogicalPlan].

Examples include:

map followed by filter Logical Plan
map followed by another map Logical Plan
groupByKey followed by agg Logical Plan

Example — `map` followed by `filter` Logical Plan



scala> spark.range(4).map(n => n * 2).filter(n => n < 3).explain(extended = true)
== Parsed Logical Plan ==
'TypedFilter <function1>, long, [StructField(value,LongType,false)], unresolveddeserializer(upcast(getcolumnbyordinal(0, LongType), LongType, - root class: "scala.Long"))
+- SerializeFromObject [input[0, bigint, true] AS value#185L]
   +- MapElements <function1>, class java.lang.Long, [StructField(value,LongType,true)], obj#184: bigint
      +- DeserializeToObject newInstance(class java.lang.Long), obj#183: java.lang.Long
         +- Range (0, 4, step=1, splits=Some(8))

== Analyzed Logical Plan ==
value: bigint
TypedFilter <function1>, long, [StructField(value,LongType,false)], cast(value#185L as bigint)
+- SerializeFromObject [input[0, bigint, true] AS value#185L]
   +- MapElements <function1>, class java.lang.Long, [StructField(value,LongType,true)], obj#184: bigint
      +- DeserializeToObject newInstance(class java.lang.Long), obj#183: java.lang.Long
         +- Range (0, 4, step=1, splits=Some(8))

== Optimized Logical Plan ==
SerializeFromObject [input[0, bigint, true] AS value#185L]
+- Filter <function1>.apply
   +- MapElements <function1>, class java.lang.Long, [StructField(value,LongType,true)], obj#184: bigint
      +- DeserializeToObject newInstance(class java.lang.Long), obj#183: java.lang.Long
         +- Range (0, 4, step=1, splits=Some(8))

== Physical Plan ==
*SerializeFromObject [input[0, bigint, true] AS value#185L]
+- *Filter <function1>.apply
   +- *MapElements <function1>, obj#184: bigint
      +- *DeserializeToObject newInstance(class java.lang.Long), obj#183: java.lang.Long
         +- *Range (0, 4, step=1, splits=Some(8))

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

scala> spark.range(4).map(n => n * 2).filter(n => n < 3).explain(extended = true)

== Parsed Logical Plan ==

'TypedFilter <function1>, long, [StructField(value,LongType,false)], unresolveddeserializer(upcast(getcolumnbyordinal(0, LongType), LongType, - root class: "scala.Long"))

+- SerializeFromObject [input[0, bigint, true] AS value#185L]

+- MapElements <function1>, class java.lang.Long, [StructField(value,LongType,true)], obj#184: bigint

+- DeserializeToObject newInstance(class java.lang.Long), obj#183: java.lang.Long

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

== Analyzed Logical Plan ==

value: bigint

TypedFilter <function1>, long, [StructField(value,LongType,false)], cast(value#185L as bigint)

+- SerializeFromObject [input[0, bigint, true] AS value#185L]

+- MapElements <function1>, class java.lang.Long, [StructField(value,LongType,true)], obj#184: bigint

+- DeserializeToObject newInstance(class java.lang.Long), obj#183: java.lang.Long

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

== Optimized Logical Plan ==

SerializeFromObject [input[0, bigint, true] AS value#185L]

+- Filter <function1>.apply

+- MapElements <function1>, class java.lang.Long, [StructField(value,LongType,true)], obj#184: bigint

+- DeserializeToObject newInstance(class java.lang.Long), obj#183: java.lang.Long

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

== Physical Plan ==

*SerializeFromObject [input[0, bigint, true] AS value#185L]

+- *Filter <function1>.apply

+- *MapElements <function1>, obj#184: bigint

+- *DeserializeToObject newInstance(class java.lang.Long), obj#183: java.lang.Long

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

Example — `map` followed by another `map` Logical Plan



// Notice unnecessary mapping between String and Int types
val query = spark.range(3).map(_.toString).map(_.toInt)

scala> query.explain(extended = true)
...
TRACE SparkOptimizer:
=== Applying Rule org.apache.spark.sql.catalyst.optimizer.EliminateSerialization ===
 SerializeFromObject [input[0, int, true] AS value#91]                                                                                                                     SerializeFromObject [input[0, int, true] AS value#91]
 +- MapElements <function1>, class java.lang.String, [StructField(value,StringType,true)], obj#90: int                                                                     +- MapElements <function1>, class java.lang.String, [StructField(value,StringType,true)], obj#90: int
!   +- DeserializeToObject value#86.toString, obj#89: java.lang.String                                                                                                        +- Project [obj#85 AS obj#89]
!      +- SerializeFromObject [staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, input[0, java.lang.String, true], true) AS value#86]         +- MapElements <function1>, class java.lang.Long, [StructField(value,LongType,true)], obj#85: java.lang.String
!         +- MapElements <function1>, class java.lang.Long, [StructField(value,LongType,true)], obj#85: java.lang.String                                                            +- DeserializeToObject newInstance(class java.lang.Long), obj#84: java.lang.Long
!            +- DeserializeToObject newInstance(class java.lang.Long), obj#84: java.lang.Long                                                                                          +- Range (0, 3, step=1, splits=Some(8))
!               +- Range (0, 3, step=1, splits=Some(8))
...
== Parsed Logical Plan ==
'SerializeFromObject [input[0, int, true] AS value#91]
+- 'MapElements <function1>, class java.lang.String, [StructField(value,StringType,true)], obj#90: int
   +- 'DeserializeToObject unresolveddeserializer(upcast(getcolumnbyordinal(0, StringType), StringType, - root class: "java.lang.String").toString), obj#89: java.lang.String
      +- SerializeFromObject [staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, input[0, java.lang.String, true], true) AS value#86]
         +- MapElements <function1>, class java.lang.Long, [StructField(value,LongType,true)], obj#85: java.lang.String
            +- DeserializeToObject newInstance(class java.lang.Long), obj#84: java.lang.Long
               +- Range (0, 3, step=1, splits=Some(8))

== Analyzed Logical Plan ==
value: int
SerializeFromObject [input[0, int, true] AS value#91]
+- MapElements <function1>, class java.lang.String, [StructField(value,StringType,true)], obj#90: int
   +- DeserializeToObject cast(value#86 as string).toString, obj#89: java.lang.String
      +- SerializeFromObject [staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, input[0, java.lang.String, true], true) AS value#86]
         +- MapElements <function1>, class java.lang.Long, [StructField(value,LongType,true)], obj#85: java.lang.String
            +- DeserializeToObject newInstance(class java.lang.Long), obj#84: java.lang.Long
               +- Range (0, 3, step=1, splits=Some(8))

== Optimized Logical Plan ==
SerializeFromObject [input[0, int, true] AS value#91]
+- MapElements <function1>, class java.lang.String, [StructField(value,StringType,true)], obj#90: int
   +- MapElements <function1>, class java.lang.Long, [StructField(value,LongType,true)], obj#85: java.lang.String
      +- DeserializeToObject newInstance(class java.lang.Long), obj#84: java.lang.Long
         +- Range (0, 3, step=1, splits=Some(8))

== Physical Plan ==
*SerializeFromObject [input[0, int, true] AS value#91]
+- *MapElements <function1>, obj#90: int
   +- *MapElements <function1>, obj#85: java.lang.String
      +- *DeserializeToObject newInstance(class java.lang.Long), obj#84: java.lang.Long
         +- *Range (0, 3, step=1, splits=Some(8))

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

// Notice unnecessary mapping between String and Int types

val query = spark.range(3).map(_.toString).map(_.toInt)

scala> query.explain(extended = true)

...

TRACE SparkOptimizer:

=== Applying Rule org.apache.spark.sql.catalyst.optimizer.EliminateSerialization ===

SerializeFromObject [input[0, int, true] AS value#91] SerializeFromObject [input[0, int, true] AS value#91]

+- MapElements <function1>, class java.lang.String, [StructField(value,StringType,true)], obj#90: int +- MapElements <function1>, class java.lang.String, [StructField(value,StringType,true)], obj#90: int

! +- DeserializeToObject value#86.toString, obj#89: java.lang.String +- Project [obj#85 AS obj#89]

! +- SerializeFromObject [staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, input[0, java.lang.String, true], true) AS value#86] +- MapElements <function1>, class java.lang.Long, [StructField(value,LongType,true)], obj#85: java.lang.String

! +- MapElements <function1>, class java.lang.Long, [StructField(value,LongType,true)], obj#85: java.lang.String +- DeserializeToObject newInstance(class java.lang.Long), obj#84: java.lang.Long

! +- DeserializeToObject newInstance(class java.lang.Long), obj#84: java.lang.Long +- Range (0, 3, step=1, splits=Some(8))

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

...

== Parsed Logical Plan ==

'SerializeFromObject [input[0, int, true] AS value#91]

+- 'MapElements <function1>, class java.lang.String, [StructField(value,StringType,true)], obj#90: int

+- 'DeserializeToObject unresolveddeserializer(upcast(getcolumnbyordinal(0, StringType), StringType, - root class: "java.lang.String").toString), obj#89: java.lang.String

+- SerializeFromObject [staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, input[0, java.lang.String, true], true) AS value#86]

+- MapElements <function1>, class java.lang.Long, [StructField(value,LongType,true)], obj#85: java.lang.String

+- DeserializeToObject newInstance(class java.lang.Long), obj#84: java.lang.Long

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

== Analyzed Logical Plan ==

value: int

SerializeFromObject [input[0, int, true] AS value#91]

+- MapElements <function1>, class java.lang.String, [StructField(value,StringType,true)], obj#90: int

+- DeserializeToObject cast(value#86 as string).toString, obj#89: java.lang.String

+- SerializeFromObject [staticinvoke(class org.apache.spark.unsafe.types.UTF8String, StringType, fromString, input[0, java.lang.String, true], true) AS value#86]

+- MapElements <function1>, class java.lang.Long, [StructField(value,LongType,true)], obj#85: java.lang.String

+- DeserializeToObject newInstance(class java.lang.Long), obj#84: java.lang.Long

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

== Optimized Logical Plan ==

SerializeFromObject [input[0, int, true] AS value#91]

+- MapElements <function1>, class java.lang.String, [StructField(value,StringType,true)], obj#90: int

+- MapElements <function1>, class java.lang.Long, [StructField(value,LongType,true)], obj#85: java.lang.String

+- DeserializeToObject newInstance(class java.lang.Long), obj#84: java.lang.Long

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

== Physical Plan ==

*SerializeFromObject [input[0, int, true] AS value#91]

+- *MapElements <function1>, obj#90: int

+- *MapElements <function1>, obj#85: java.lang.String

+- *DeserializeToObject newInstance(class java.lang.Long), obj#84: java.lang.Long

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

Example — `groupByKey` followed by `agg` Logical Plan



scala> spark.range(4).map(n => (n, n % 2)).groupByKey(_._2).agg(typed.sum(_._2)).explain(true)
== Parsed Logical Plan ==
'Aggregate [value#454L], [value#454L, unresolvedalias(typedsumdouble(org.apache.spark.sql.execution.aggregate.TypedSumDouble@4fcb0de4, Some(unresolveddeserializer(newInstance(class scala.Tuple2), _1#450L, _2#451L)), Some(class scala.Tuple2), Some(StructType(StructField(_1,LongType,true), StructField(_2,LongType,false))), input[0, double, true] AS value#457, unresolveddeserializer(upcast(getcolumnbyordinal(0, DoubleType), DoubleType, - root class: "scala.Double"), value#457), input[0, double, true] AS value#456, DoubleType, DoubleType, false), Some(<function1>))]
+- AppendColumns <function1>, class scala.Tuple2, [StructField(_1,LongType,true), StructField(_2,LongType,false)], newInstance(class scala.Tuple2), [input[0, bigint, true] AS value#454L]
   +- SerializeFromObject [assertnotnull(input[0, scala.Tuple2, true], top level non-flat input object)._1.longValue AS _1#450L, assertnotnull(input[0, scala.Tuple2, true], top level non-flat input object)._2 AS _2#451L]
      +- MapElements <function1>, class java.lang.Long, [StructField(value,LongType,true)], obj#449: scala.Tuple2
         +- DeserializeToObject newInstance(class java.lang.Long), obj#448: java.lang.Long
            +- Range (0, 4, step=1, splits=Some(8))

== Analyzed Logical Plan ==
value: bigint, TypedSumDouble(scala.Tuple2): double
Aggregate [value#454L], [value#454L, typedsumdouble(org.apache.spark.sql.execution.aggregate.TypedSumDouble@4fcb0de4, Some(newInstance(class scala.Tuple2)), Some(class scala.Tuple2), Some(StructType(StructField(_1,LongType,true), StructField(_2,LongType,false))), input[0, double, true] AS value#457, cast(value#457 as double), input[0, double, true] AS value#456, DoubleType, DoubleType, false) AS TypedSumDouble(scala.Tuple2)#462]
+- AppendColumns <function1>, class scala.Tuple2, [StructField(_1,LongType,true), StructField(_2,LongType,false)], newInstance(class scala.Tuple2), [input[0, bigint, true] AS value#454L]
   +- SerializeFromObject [assertnotnull(input[0, scala.Tuple2, true], top level non-flat input object)._1.longValue AS _1#450L, assertnotnull(input[0, scala.Tuple2, true], top level non-flat input object)._2 AS _2#451L]
      +- MapElements <function1>, class java.lang.Long, [StructField(value,LongType,true)], obj#449: scala.Tuple2
         +- DeserializeToObject newInstance(class java.lang.Long), obj#448: java.lang.Long
            +- Range (0, 4, step=1, splits=Some(8))

== Optimized Logical Plan ==
Aggregate [value#454L], [value#454L, typedsumdouble(org.apache.spark.sql.execution.aggregate.TypedSumDouble@4fcb0de4, Some(newInstance(class scala.Tuple2)), Some(class scala.Tuple2), Some(StructType(StructField(_1,LongType,true), StructField(_2,LongType,false))), input[0, double, true] AS value#457, value#457, input[0, double, true] AS value#456, DoubleType, DoubleType, false) AS TypedSumDouble(scala.Tuple2)#462]
+- AppendColumnsWithObject <function1>, [assertnotnull(input[0, scala.Tuple2, true], top level non-flat input object)._1.longValue AS _1#450L, assertnotnull(input[0, scala.Tuple2, true], top level non-flat input object)._2 AS _2#451L], [input[0, bigint, true] AS value#454L]
   +- MapElements <function1>, class java.lang.Long, [StructField(value,LongType,true)], obj#449: scala.Tuple2
      +- DeserializeToObject newInstance(class java.lang.Long), obj#448: java.lang.Long
         +- Range (0, 4, step=1, splits=Some(8))

== Physical Plan ==
*HashAggregate(keys=[value#454L], functions=[typedsumdouble(org.apache.spark.sql.execution.aggregate.TypedSumDouble@4fcb0de4, Some(newInstance(class scala.Tuple2)), Some(class scala.Tuple2), Some(StructType(StructField(_1,LongType,true), StructField(_2,LongType,false))), input[0, double, true] AS value#457, value#457, input[0, double, true] AS value#456, DoubleType, DoubleType, false)], output=[value#454L, TypedSumDouble(scala.Tuple2)#462])
+- Exchange hashpartitioning(value#454L, 200)
   +- *HashAggregate(keys=[value#454L], functions=[partial_typedsumdouble(org.apache.spark.sql.execution.aggregate.TypedSumDouble@4fcb0de4, Some(newInstance(class scala.Tuple2)), Some(class scala.Tuple2), Some(StructType(StructField(_1,LongType,true), StructField(_2,LongType,false))), input[0, double, true] AS value#457, value#457, input[0, double, true] AS value#456, DoubleType, DoubleType, false)], output=[value#454L, value#463])
      +- AppendColumnsWithObject <function1>, [assertnotnull(input[0, scala.Tuple2, true], top level non-flat input object)._1.longValue AS _1#450L, assertnotnull(input[0, scala.Tuple2, true], top level non-flat input object)._2 AS _2#451L], [input[0, bigint, true] AS value#454L]
         +- MapElements <function1>, obj#449: scala.Tuple2
            +- DeserializeToObject newInstance(class java.lang.Long), obj#448: java.lang.Long
               +- *Range (0, 4, step=1, splits=Some(8))

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

scala> spark.range(4).map(n => (n, n % 2)).groupByKey(_._2).agg(typed.sum(_._2)).explain(true)

== Parsed Logical Plan ==

'Aggregate [value#454L], [value#454L, unresolvedalias(typedsumdouble(org.apache.spark.sql.execution.aggregate.TypedSumDouble@4fcb0de4, Some(unresolveddeserializer(newInstance(class scala.Tuple2), _1#450L, _2#451L)), Some(class scala.Tuple2), Some(StructType(StructField(_1,LongType,true), StructField(_2,LongType,false))), input[0, double, true] AS value#457, unresolveddeserializer(upcast(getcolumnbyordinal(0, DoubleType), DoubleType, - root class: "scala.Double"), value#457), input[0, double, true] AS value#456, DoubleType, DoubleType, false), Some(<function1>))]

+- AppendColumns <function1>, class scala.Tuple2, [StructField(_1,LongType,true), StructField(_2,LongType,false)], newInstance(class scala.Tuple2), [input[0, bigint, true] AS value#454L]

+- SerializeFromObject [assertnotnull(input[0, scala.Tuple2, true], top level non-flat input object)._1.longValue AS _1#450L, assertnotnull(input[0, scala.Tuple2, true], top level non-flat input object)._2 AS _2#451L]

+- MapElements <function1>, class java.lang.Long, [StructField(value,LongType,true)], obj#449: scala.Tuple2

+- DeserializeToObject newInstance(class java.lang.Long), obj#448: java.lang.Long

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

== Analyzed Logical Plan ==

value: bigint, TypedSumDouble(scala.Tuple2): double

Aggregate [value#454L], [value#454L, typedsumdouble(org.apache.spark.sql.execution.aggregate.TypedSumDouble@4fcb0de4, Some(newInstance(class scala.Tuple2)), Some(class scala.Tuple2), Some(StructType(StructField(_1,LongType,true), StructField(_2,LongType,false))), input[0, double, true] AS value#457, cast(value#457 as double), input[0, double, true] AS value#456, DoubleType, DoubleType, false) AS TypedSumDouble(scala.Tuple2)#462]

+- AppendColumns <function1>, class scala.Tuple2, [StructField(_1,LongType,true), StructField(_2,LongType,false)], newInstance(class scala.Tuple2), [input[0, bigint, true] AS value#454L]

+- SerializeFromObject [assertnotnull(input[0, scala.Tuple2, true], top level non-flat input object)._1.longValue AS _1#450L, assertnotnull(input[0, scala.Tuple2, true], top level non-flat input object)._2 AS _2#451L]

+- MapElements <function1>, class java.lang.Long, [StructField(value,LongType,true)], obj#449: scala.Tuple2

+- DeserializeToObject newInstance(class java.lang.Long), obj#448: java.lang.Long

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

== Optimized Logical Plan ==

Aggregate [value#454L], [value#454L, typedsumdouble(org.apache.spark.sql.execution.aggregate.TypedSumDouble@4fcb0de4, Some(newInstance(class scala.Tuple2)), Some(class scala.Tuple2), Some(StructType(StructField(_1,LongType,true), StructField(_2,LongType,false))), input[0, double, true] AS value#457, value#457, input[0, double, true] AS value#456, DoubleType, DoubleType, false) AS TypedSumDouble(scala.Tuple2)#462]

+- AppendColumnsWithObject <function1>, [assertnotnull(input[0, scala.Tuple2, true], top level non-flat input object)._1.longValue AS _1#450L, assertnotnull(input[0, scala.Tuple2, true], top level non-flat input object)._2 AS _2#451L], [input[0, bigint, true] AS value#454L]

+- MapElements <function1>, class java.lang.Long, [StructField(value,LongType,true)], obj#449: scala.Tuple2

+- DeserializeToObject newInstance(class java.lang.Long), obj#448: java.lang.Long

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

== Physical Plan ==

*HashAggregate(keys=[value#454L], functions=[typedsumdouble(org.apache.spark.sql.execution.aggregate.TypedSumDouble@4fcb0de4, Some(newInstance(class scala.Tuple2)), Some(class scala.Tuple2), Some(StructType(StructField(_1,LongType,true), StructField(_2,LongType,false))), input[0, double, true] AS value#457, value#457, input[0, double, true] AS value#456, DoubleType, DoubleType, false)], output=[value#454L, TypedSumDouble(scala.Tuple2)#462])

+- Exchange hashpartitioning(value#454L, 200)

+- *HashAggregate(keys=[value#454L], functions=[partial_typedsumdouble(org.apache.spark.sql.execution.aggregate.TypedSumDouble@4fcb0de4, Some(newInstance(class scala.Tuple2)), Some(class scala.Tuple2), Some(StructType(StructField(_1,LongType,true), StructField(_2,LongType,false))), input[0, double, true] AS value#457, value#457, input[0, double, true] AS value#456, DoubleType, DoubleType, false)], output=[value#454L, value#463])

+- AppendColumnsWithObject <function1>, [assertnotnull(input[0, scala.Tuple2, true], top level non-flat input object)._1.longValue AS _1#450L, assertnotnull(input[0, scala.Tuple2, true], top level non-flat input object)._2 AS _2#451L], [input[0, bigint, true] AS value#454L]

+- MapElements <function1>, obj#449: scala.Tuple2

+- DeserializeToObject newInstance(class java.lang.Long), obj#448: java.lang.Long

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

Executing Rule — `apply` Method



apply(plan: LogicalPlan): LogicalPlan

1

2

3

4

5

apply(plan: LogicalPlan): LogicalPlan

Note	`apply` is part of the Rule Contract to execute (apply) a rule on a TreeNode (e.g. LogicalPlan).

apply…FIXME

DecimalAggregates

2013-03-20admin阅读(1572)

DecimalAggregates Logical Optimization

DecimalAggregates is a base logical optimization that transforms Sum and Average aggregate functions on fixed-precision DecimalType values to use UnscaledValue (unscaled Long) values in WindowExpression and AggregateExpression expressions.

DecimalAggregates is part of the Decimal Optimizations fixed-point batch in the standard batches of the Catalyst Optimizer.

DecimalAggregates is simply a Catalyst rule for transforming logical plans, i.e. Rule[LogicalPlan].

Tip

Import DecimalAggregates and apply the rule directly on your structured queries to learn how the rule works.



import org.apache.spark.sql.catalyst.optimizer.DecimalAggregates
val da = DecimalAggregates(spark.sessionState.conf)

// Build analyzed logical plan
// with sum aggregate function and Decimal field
import org.apache.spark.sql.types.DecimalType
val query = spark.range(5).select(sum($"id" cast DecimalType(1,0)) as "sum")
scala> val plan = query.queryExecution.analyzed
plan: org.apache.spark.sql.catalyst.plans.logical.LogicalPlan =
Aggregate [sum(cast(id#91L as decimal(1,0))) AS sum#95]
+- Range (0, 5, step=1, splits=Some(8))

// Apply DecimalAggregates rule
// Note MakeDecimal and UnscaledValue operators
scala> da.apply(plan)
res27: org.apache.spark.sql.catalyst.plans.logical.LogicalPlan =
Aggregate [MakeDecimal(sum(UnscaledValue(cast(id#91L as decimal(1,0)))),11,0) AS sum#95]
+- Range (0, 5, step=1, splits=Some(8))

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

import org.apache.spark.sql.catalyst.optimizer.DecimalAggregates

val da = DecimalAggregates(spark.sessionState.conf)

// Build analyzed logical plan

// with sum aggregate function and Decimal field

import org.apache.spark.sql.types.DecimalType

val query = spark.range(5).select(sum($"id" cast DecimalType(1,0)) as "sum")

scala> val plan = query.queryExecution.analyzed

plan: org.apache.spark.sql.catalyst.plans.logical.LogicalPlan =

Aggregate [sum(cast(id#91L as decimal(1,0))) AS sum#95]

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

// Apply DecimalAggregates rule

// Note MakeDecimal and UnscaledValue operators

scala> da.apply(plan)

res27: org.apache.spark.sql.catalyst.plans.logical.LogicalPlan =

Aggregate [MakeDecimal(sum(UnscaledValue(cast(id#91L as decimal(1,0)))),11,0) AS sum#95]

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

Example: sum Aggregate Function on Decimal with Precision Smaller Than 9



// sum aggregate with Decimal field with precision <= 8
val q = "SELECT sum(cast(id AS DECIMAL(5,0))) FROM range(1)"

scala> sql(q).explain(true)
== Parsed Logical Plan ==
'Project [unresolvedalias('sum(cast('id as decimal(5,0))), None)]
+- 'UnresolvedTableValuedFunction range, [1]

== Analyzed Logical Plan ==
sum(CAST(id AS DECIMAL(5,0))): decimal(15,0)
Aggregate [sum(cast(id#104L as decimal(5,0))) AS sum(CAST(id AS DECIMAL(5,0)))#106]
+- Range (0, 1, step=1, splits=None)

== Optimized Logical Plan ==
Aggregate [MakeDecimal(sum(UnscaledValue(cast(id#104L as decimal(5,0)))),15,0) AS sum(CAST(id AS DECIMAL(5,0)))#106]
+- Range (0, 1, step=1, splits=None)

== Physical Plan ==
*HashAggregate(keys=[], functions=[sum(UnscaledValue(cast(id#104L as decimal(5,0))))], output=[sum(CAST(id AS DECIMAL(5,0)))#106])
+- Exchange SinglePartition
   +- *HashAggregate(keys=[], functions=[partial_sum(UnscaledValue(cast(id#104L as decimal(5,0))))], output=[sum#108L])
      +- *Range (0, 1, step=1, splits=None)

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

// sum aggregate with Decimal field with precision <= 8

val q = "SELECT sum(cast(id AS DECIMAL(5,0))) FROM range(1)"

scala> sql(q).explain(true)

== Parsed Logical Plan ==

'Project [unresolvedalias('sum(cast('id as decimal(5,0))), None)]

+- 'UnresolvedTableValuedFunction range, [1]

== Analyzed Logical Plan ==

sum(CAST(id AS DECIMAL(5,0))): decimal(15,0)

Aggregate [sum(cast(id#104L as decimal(5,0))) AS sum(CAST(id AS DECIMAL(5,0)))#106]

+- Range (0, 1, step=1, splits=None)

== Optimized Logical Plan ==

Aggregate [MakeDecimal(sum(UnscaledValue(cast(id#104L as decimal(5,0)))),15,0) AS sum(CAST(id AS DECIMAL(5,0)))#106]

+- Range (0, 1, step=1, splits=None)

== Physical Plan ==

*HashAggregate(keys=[], functions=[sum(UnscaledValue(cast(id#104L as decimal(5,0))))], output=[sum(CAST(id AS DECIMAL(5,0)))#106])

+- Exchange SinglePartition

+- *HashAggregate(keys=[], functions=[partial_sum(UnscaledValue(cast(id#104L as decimal(5,0))))], output=[sum#108L])

+- *Range (0, 1, step=1, splits=None)

Example: avg Aggregate Function on Decimal with Precision Smaller Than 12



// avg aggregate with Decimal field with precision <= 11
val q = "SELECT avg(cast(id AS DECIMAL(10,0))) FROM range(1)"

scala> val q = "SELECT avg(cast(id AS DECIMAL(10,0))) FROM range(1)"
q: String = SELECT avg(cast(id AS DECIMAL(10,0))) FROM range(1)

scala> sql(q).explain(true)
== Parsed Logical Plan ==
'Project [unresolvedalias('avg(cast('id as decimal(10,0))), None)]
+- 'UnresolvedTableValuedFunction range, [1]

== Analyzed Logical Plan ==
avg(CAST(id AS DECIMAL(10,0))): decimal(14,4)
Aggregate [avg(cast(id#115L as decimal(10,0))) AS avg(CAST(id AS DECIMAL(10,0)))#117]
+- Range (0, 1, step=1, splits=None)

== Optimized Logical Plan ==
Aggregate [cast((avg(UnscaledValue(cast(id#115L as decimal(10,0)))) / 1.0) as decimal(14,4)) AS avg(CAST(id AS DECIMAL(10,0)))#117]
+- Range (0, 1, step=1, splits=None)

== Physical Plan ==
*HashAggregate(keys=[], functions=[avg(UnscaledValue(cast(id#115L as decimal(10,0))))], output=[avg(CAST(id AS DECIMAL(10,0)))#117])
+- Exchange SinglePartition
   +- *HashAggregate(keys=[], functions=[partial_avg(UnscaledValue(cast(id#115L as decimal(10,0))))], output=[sum#120, count#121L])
      +- *Range (0, 1, step=1, splits=None)

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

// avg aggregate with Decimal field with precision <= 11

val q = "SELECT avg(cast(id AS DECIMAL(10,0))) FROM range(1)"

scala> val q = "SELECT avg(cast(id AS DECIMAL(10,0))) FROM range(1)"

q: String = SELECT avg(cast(id AS DECIMAL(10,0))) FROM range(1)

scala> sql(q).explain(true)

== Parsed Logical Plan ==

'Project [unresolvedalias('avg(cast('id as decimal(10,0))), None)]

+- 'UnresolvedTableValuedFunction range, [1]

== Analyzed Logical Plan ==

avg(CAST(id AS DECIMAL(10,0))): decimal(14,4)

Aggregate [avg(cast(id#115L as decimal(10,0))) AS avg(CAST(id AS DECIMAL(10,0)))#117]

+- Range (0, 1, step=1, splits=None)

== Optimized Logical Plan ==

Aggregate [cast((avg(UnscaledValue(cast(id#115L as decimal(10,0)))) / 1.0) as decimal(14,4)) AS avg(CAST(id AS DECIMAL(10,0)))#117]

+- Range (0, 1, step=1, splits=None)

== Physical Plan ==

*HashAggregate(keys=[], functions=[avg(UnscaledValue(cast(id#115L as decimal(10,0))))], output=[avg(CAST(id AS DECIMAL(10,0)))#117])

+- Exchange SinglePartition

+- *HashAggregate(keys=[], functions=[partial_avg(UnscaledValue(cast(id#115L as decimal(10,0))))], output=[sum#120, count#121L])

+- *Range (0, 1, step=1, splits=None)

Executing Rule — `apply` Method



apply(plan: LogicalPlan): LogicalPlan

1

2

3

4

5

apply(plan: LogicalPlan): LogicalPlan

Note	`apply` is part of the Rule Contract to execute (apply) a rule on a TreeNode (e.g. LogicalPlan).

apply…FIXME

CostBasedJoinReorder

2013-03-19admin阅读(1345)

CostBasedJoinReorder Logical Optimization — Join Reordering in Cost-Based Optimization

CostBasedJoinReorder is a base logical optimization that reorders joins in cost-based optimization.

ReorderJoin is part of the Join Reorder once-executed batch in the standard batches of the Catalyst Optimizer.

ReorderJoin is simply a Catalyst rule for transforming logical plans, i.e. Rule[LogicalPlan].

CostBasedJoinReorder applies the join optimizations on a logical plan with 2 or more consecutive inner or cross joins (possibly separated by Project operators) when spark.sql.cbo.enabled and spark.sql.cbo.joinReorder.enabled configuration properties are both enabled.



// Use shortcuts to read the values of the properties
scala> spark.sessionState.conf.cboEnabled
res0: Boolean = true

scala> spark.sessionState.conf.joinReorderEnabled
res1: Boolean = true

1

2

3

4

5

6

7

8

9

10

// Use shortcuts to read the values of the properties

scala> spark.sessionState.conf.cboEnabled

res0: Boolean = true

scala> spark.sessionState.conf.joinReorderEnabled

res1: Boolean = true

CostBasedJoinReorder uses row count statistic that is computed using ANALYZE TABLE COMPUTE STATISTICS SQL command with no NOSCAN option.



// Create tables and compute their row count statistics
// There have to be at least 2 joins
// Make the example reproducible
val tableNames = Seq("t1", "t2", "tiny")
import org.apache.spark.sql.catalyst.TableIdentifier
val tableIds = tableNames.map(TableIdentifier.apply)
val sessionCatalog = spark.sessionState.catalog
tableIds.foreach { tableId =>
  sessionCatalog.dropTable(tableId, ignoreIfNotExists = true, purge = true)
}

val belowBroadcastJoinThreshold = spark.sessionState.conf.autoBroadcastJoinThreshold - 1
spark.range(belowBroadcastJoinThreshold).write.saveAsTable("t1")
// t2 is twice as big as t1
spark.range(2 * belowBroadcastJoinThreshold).write.saveAsTable("t2")
spark.range(5).write.saveAsTable("tiny")

// Compute row count statistics
tableNames.foreach { t =>
  sql(s"ANALYZE TABLE $t COMPUTE STATISTICS")
}

// Load the tables
val t1 = spark.table("t1")
val t2 = spark.table("t2")
val tiny = spark.table("tiny")

// Example: Inner join with join condition
val q = t1.join(t2, Seq("id")).join(tiny, Seq("id"))
val plan = q.queryExecution.analyzed
scala> println(plan.numberedTreeString)
00 Project [id#51L]
01 +- Join Inner, (id#51L = id#57L)
02    :- Project [id#51L]
03    :  +- Join Inner, (id#51L = id#54L)
04    :     :- SubqueryAlias t1
05    :     :  +- Relation[id#51L] parquet
06    :     +- SubqueryAlias t2
07    :        +- Relation[id#54L] parquet
08    +- SubqueryAlias tiny
09       +- Relation[id#57L] parquet

// Eliminate SubqueryAlias logical operators as they no longer needed
// And "confuse" CostBasedJoinReorder
// CostBasedJoinReorder cares about how deep Joins are and reorders consecutive joins only
import org.apache.spark.sql.catalyst.analysis.EliminateSubqueryAliases
val noAliasesPlan = EliminateSubqueryAliases(plan)
scala> println(noAliasesPlan.numberedTreeString)
00 Project [id#51L]
01 +- Join Inner, (id#51L = id#57L)
02    :- Project [id#51L]
03    :  +- Join Inner, (id#51L = id#54L)
04    :     :- Relation[id#51L] parquet
05    :     +- Relation[id#54L] parquet
06    +- Relation[id#57L] parquet

// Let's go pro and create a custom RuleExecutor (i.e. an Optimizer)
import org.apache.spark.sql.catalyst.rules.RuleExecutor
import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
import org.apache.spark.sql.catalyst.analysis.EliminateSubqueryAliases
import org.apache.spark.sql.catalyst.optimizer.CostBasedJoinReorder
object Optimize extends RuleExecutor[LogicalPlan] {
  val batches =
    Batch("EliminateSubqueryAliases", Once, EliminateSubqueryAliases) ::
    Batch("Join Reorder", Once, CostBasedJoinReorder) :: Nil
}

val joinsReordered = Optimize.execute(plan)
scala> println(joinsReordered.numberedTreeString)
00 Project [id#51L]
01 +- Join Inner, (id#51L = id#54L)
02    :- Project [id#51L]
03    :  +- Join Inner, (id#51L = id#57L)
04    :     :- Relation[id#51L] parquet
05    :     +- Relation[id#57L] parquet
06    +- Relation[id#54L] parquet

// Execute the plans
// Compare the plans as diagrams in web UI @ http://localhost:4040/SQL
// We'd have to use too many internals so let's turn CBO on and off
// Moreover, please remember that the query "phases" are cached
// That's why we copy and paste the entire query for execution
import org.apache.spark.sql.internal.SQLConf
val cc = SQLConf.get
cc.setConf(SQLConf.CBO_ENABLED, false)
val q = t1.join(t2, Seq("id")).join(tiny, Seq("id"))
q.collect.foreach(_ => ())

cc.setConf(SQLConf.CBO_ENABLED, true)
val q = t1.join(t2, Seq("id")).join(tiny, Seq("id"))
q.collect.foreach(_ => ())

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

78

79

80

81

82

83

84

85

86

87

88

89

90

91

92

93

94

95

// Create tables and compute their row count statistics

// There have to be at least 2 joins

// Make the example reproducible

val tableNames = Seq("t1", "t2", "tiny")

import org.apache.spark.sql.catalyst.TableIdentifier

val tableIds = tableNames.map(TableIdentifier.apply)

val sessionCatalog = spark.sessionState.catalog

tableIds.foreach { tableId =>

sessionCatalog.dropTable(tableId, ignoreIfNotExists = true, purge = true)

}

val belowBroadcastJoinThreshold = spark.sessionState.conf.autoBroadcastJoinThreshold - 1

spark.range(belowBroadcastJoinThreshold).write.saveAsTable("t1")

// t2 is twice as big as t1

spark.range(2 * belowBroadcastJoinThreshold).write.saveAsTable("t2")

spark.range(5).write.saveAsTable("tiny")

// Compute row count statistics

tableNames.foreach { t =>

sql(s"ANALYZE TABLE $t COMPUTE STATISTICS")

}

// Load the tables

val t1 = spark.table("t1")

val t2 = spark.table("t2")

val tiny = spark.table("tiny")

// Example: Inner join with join condition

val q = t1.join(t2, Seq("id")).join(tiny, Seq("id"))

val plan = q.queryExecution.analyzed

scala> println(plan.numberedTreeString)

00 Project [id#51L]

01 +- Join Inner, (id#51L = id#57L)

02 :- Project [id#51L]

03 : +- Join Inner, (id#51L = id#54L)

04 : :- SubqueryAlias t1

05 : : +- Relation[id#51L] parquet

06 : +- SubqueryAlias t2

07 : +- Relation[id#54L] parquet

08 +- SubqueryAlias tiny

09 +- Relation[id#57L] parquet

// Eliminate SubqueryAlias logical operators as they no longer needed

// And "confuse" CostBasedJoinReorder

// CostBasedJoinReorder cares about how deep Joins are and reorders consecutive joins only

import org.apache.spark.sql.catalyst.analysis.EliminateSubqueryAliases

val noAliasesPlan = EliminateSubqueryAliases(plan)

scala> println(noAliasesPlan.numberedTreeString)

00 Project [id#51L]

01 +- Join Inner, (id#51L = id#57L)

02 :- Project [id#51L]

03 : +- Join Inner, (id#51L = id#54L)

04 : :- Relation[id#51L] parquet

05 : +- Relation[id#54L] parquet

06 +- Relation[id#57L] parquet

// Let's go pro and create a custom RuleExecutor (i.e. an Optimizer)

import org.apache.spark.sql.catalyst.rules.RuleExecutor

import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan

import org.apache.spark.sql.catalyst.analysis.EliminateSubqueryAliases

import org.apache.spark.sql.catalyst.optimizer.CostBasedJoinReorder

object Optimize extends RuleExecutor[LogicalPlan] {

val batches =

Batch("EliminateSubqueryAliases", Once, EliminateSubqueryAliases) ::

Batch("Join Reorder", Once, CostBasedJoinReorder) :: Nil

}

val joinsReordered = Optimize.execute(plan)

scala> println(joinsReordered.numberedTreeString)

00 Project [id#51L]

01 +- Join Inner, (id#51L = id#54L)

02 :- Project [id#51L]

03 : +- Join Inner, (id#51L = id#57L)

04 : :- Relation[id#51L] parquet

05 : +- Relation[id#57L] parquet

06 +- Relation[id#54L] parquet

// Execute the plans

// Compare the plans as diagrams in web UI @ http://localhost:4040/SQL

// We'd have to use too many internals so let's turn CBO on and off

// Moreover, please remember that the query "phases" are cached

// That's why we copy and paste the entire query for execution

import org.apache.spark.sql.internal.SQLConf

val cc = SQLConf.get

cc.setConf(SQLConf.CBO_ENABLED, false)

val q = t1.join(t2, Seq("id")).join(tiny, Seq("id"))

q.collect.foreach(_ => ())

cc.setConf(SQLConf.CBO_ENABLED, true)

val q = t1.join(t2, Seq("id")).join(tiny, Seq("id"))

q.collect.foreach(_ => ())

Caution

FIXME Examples of other join queries

Cross join with join condition
Project with attributes only and Inner join with join condition
Project with attributes only and Cross join with join condition

Tip

Enable DEBUG logging level for org.apache.spark.sql.catalyst.optimizer.JoinReorderDP logger to see the join reordering duration.

Add the following line to conf/log4j.properties:



log4j.logger.org.apache.spark.sql.catalyst.optimizer.JoinReorderDP=DEBUG

1

2

3

4

5

log4j.logger.org.apache.spark.sql.catalyst.optimizer.JoinReorderDP=DEBUG

Refer to Logging.

Executing Rule — `apply` Method



apply(plan: LogicalPlan): LogicalPlan

1

2

3

4

5

apply(plan: LogicalPlan): LogicalPlan

Note	`apply` is part of the Rule Contract to execute (apply) a rule on a TreeNode (e.g. LogicalPlan).

apply traverses the input logical plan down and tries to reorder the following logical operators:

Join for CROSS or INNER joins with a join condition
Project with the above Join child operator and the project list of Attribute leaf expressions only

Reordering Logical Plan with Join Operators — `reorder` Internal Method



reorder(plan: LogicalPlan, output: Seq[Attribute]): LogicalPlan

1

2

3

4

5

reorder(plan: LogicalPlan, output: Seq[Attribute]): LogicalPlan

reorder…FIXME

Note	`reorder` is used exclusively when `CostBasedJoinReorder` is applied to a logical plan.

`replaceWithOrderedJoin` Internal Method



replaceWithOrderedJoin(plan: LogicalPlan): LogicalPlan

1

2

3

4

5

replaceWithOrderedJoin(plan: LogicalPlan): LogicalPlan

replaceWithOrderedJoin…FIXME

Note	`replaceWithOrderedJoin` is used recursively and when `CostBasedJoinReorder` is reordering…FIXME

Extracting Consecutive Join Operators — `extractInnerJoins` Internal Method



extractInnerJoins(plan: LogicalPlan): (Seq[LogicalPlan], Set[Expression])

1

2

3

4

5

extractInnerJoins(plan: LogicalPlan): (Seq[LogicalPlan], Set[Expression])

extractInnerJoins finds consecutive Join logical operators (inner or cross) with join conditions or Project logical operators with Join logical operator and the project list of Attribute leaf expressions only.

For Project operators extractInnerJoins calls itself recursively with the Join operator inside.

In the end, extractInnerJoins gives the collection of logical plans under the consecutive Join logical operators (possibly separated by Project operators only) and their join conditions (for which And expressions have been split).

Note	`extractInnerJoins` is used recursively when `CostBasedJoinReorder` is reordering a logical plan.

ConstantFolding

2013-03-18admin阅读(974)

ConstantFolding Logical Optimization

ConstantFolding is a base logical optimization that replaces expressions that can be statically evaluated with their equivalent literal values.

ConstantFolding is part of the Operator Optimization before Inferring Filters fixed-point batch in the standard batches of the Catalyst Optimizer.

ConstantFolding is simply a Catalyst rule for transforming logical plans, i.e. Rule[LogicalPlan].



scala> spark.range(1).select(lit(3) > 2).explain(true)
...
TRACE SparkOptimizer:
=== Applying Rule org.apache.spark.sql.catalyst.optimizer.ConstantFolding ===
!Project [(3 > 2) AS (3 > 2)#3]            Project [true AS (3 > 2)#3]
 +- Range (0, 1, step=1, splits=Some(8))   +- Range (0, 1, step=1, splits=Some(8))

scala> spark.range(1).select('id + 'id > 0).explain(true)
...
TRACE SparkOptimizer:
=== Applying Rule org.apache.spark.sql.catalyst.optimizer.ConstantFolding ===
!Project [((id#7L + id#7L) > cast(0 as bigint)) AS ((id + id) > 0)#10]   Project [((id#7L + id#7L) > 0) AS ((id + id) > 0)#10]
 +- Range (0, 1, step=1, splits=Some(8))                                 +- Range (0, 1, step=1, splits=Some(8))

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

scala> spark.range(1).select(lit(3) > 2).explain(true)

...

TRACE SparkOptimizer:

=== Applying Rule org.apache.spark.sql.catalyst.optimizer.ConstantFolding ===

!Project [(3 > 2) AS (3 > 2)#3] Project [true AS (3 > 2)#3]

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

scala> spark.range(1).select('id + 'id > 0).explain(true)

...

TRACE SparkOptimizer:

=== Applying Rule org.apache.spark.sql.catalyst.optimizer.ConstantFolding ===

!Project [((id#7L + id#7L) > cast(0 as bigint)) AS ((id + id) > 0)#10] Project [((id#7L + id#7L) > 0) AS ((id + id) > 0)#10]

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

Executing Rule — `apply` Method



apply(plan: LogicalPlan): LogicalPlan

1

2

3

4

5

apply(plan: LogicalPlan): LogicalPlan

Note	`apply` is part of the Rule Contract to execute (apply) a rule on a TreeNode (e.g. LogicalPlan).

apply…FIXME

ComputeCurrentTime

2013-03-17admin阅读(1409)

ComputeCurrentTime Logical Optimization

ComputeCurrentTime is a base logical optimization that computes the current date and timestamp.

ComputeCurrentTime is part of the Finish Analysis once-executed batch in the standard batches of the Catalyst Optimizer.

ComputeCurrentTime is simply a Catalyst rule for transforming logical plans, i.e. Rule[LogicalPlan].



// Query with two current_date's
import org.apache.spark.sql.functions.current_date
val q = spark.range(1).select(current_date() as "d1", current_date() as "d2")
val analyzedPlan = q.queryExecution.analyzed

scala> println(analyzedPlan.numberedTreeString)
00 Project [current_date(Some(Europe/Warsaw)) AS d1#12, current_date(Some(Europe/Warsaw)) AS d2#13]
01 +- Range (0, 1, step=1, splits=Some(8))

import org.apache.spark.sql.catalyst.optimizer.ComputeCurrentTime

val afterComputeCurrentTime = ComputeCurrentTime(analyzedPlan)
scala> println(afterComputeCurrentTime.numberedTreeString)
00 Project [17773 AS d1#12, 17773 AS d2#13]
01 +- Range (0, 1, step=1, splits=Some(8))

// Another query with two current_timestamp's
// Here the millis play a bigger role so it is easier to notice the results
import org.apache.spark.sql.functions.current_timestamp
val q = spark.range(1).select(current_timestamp() as "ts1", current_timestamp() as "ts2")
val analyzedPlan = q.queryExecution.analyzed
val afterComputeCurrentTime = ComputeCurrentTime(analyzedPlan)
scala> println(afterComputeCurrentTime.numberedTreeString)
00 Project [1535629687768000 AS ts1#18, 1535629687768000 AS ts2#19]
01 +- Range (0, 1, step=1, splits=Some(8))

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

// Query with two current_date's

import org.apache.spark.sql.functions.current_date

val q = spark.range(1).select(current_date() as "d1", current_date() as "d2")

val analyzedPlan = q.queryExecution.analyzed

scala> println(analyzedPlan.numberedTreeString)

00 Project [current_date(Some(Europe/Warsaw)) AS d1#12, current_date(Some(Europe/Warsaw)) AS d2#13]

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

import org.apache.spark.sql.catalyst.optimizer.ComputeCurrentTime

val afterComputeCurrentTime = ComputeCurrentTime(analyzedPlan)

scala> println(afterComputeCurrentTime.numberedTreeString)

00 Project [17773 AS d1#12, 17773 AS d2#13]

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

// Another query with two current_timestamp's

// Here the millis play a bigger role so it is easier to notice the results

import org.apache.spark.sql.functions.current_timestamp

val q = spark.range(1).select(current_timestamp() as "ts1", current_timestamp() as "ts2")

val analyzedPlan = q.queryExecution.analyzed

val afterComputeCurrentTime = ComputeCurrentTime(analyzedPlan)

scala> println(afterComputeCurrentTime.numberedTreeString)

00 Project [1535629687768000 AS ts1#18, 1535629687768000 AS ts2#19]

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

Executing Rule — `apply` Method



apply(plan: LogicalPlan): LogicalPlan

1

2

3

4

5

apply(plan: LogicalPlan): LogicalPlan

Note	`apply` is part of the Rule Contract to execute (apply) a rule on a TreeNode (e.g. LogicalPlan).

apply…FIXME

CombineUnions

2013-03-16admin阅读(1545)

CombineUnions Logical Optimization

CombineUnions is a base logical optimization that FIXME.

CombineUnions is part of the Union once-executed batch in the standard batches of the Catalyst Optimizer.

CombineUnions is simply a Catalyst rule for transforming logical plans, i.e. Rule[LogicalPlan].



// FIXME Demo

1

2

3

4

5

// FIXME Demo

Executing Rule — `apply` Method



apply(plan: LogicalPlan): LogicalPlan

1

2

3

4

5

apply(plan: LogicalPlan): LogicalPlan

Note	`apply` is part of the Rule Contract to execute (apply) a rule on a TreeNode (e.g. LogicalPlan).

apply…FIXME

CombineTypedFilters

2013-03-15admin阅读(2255)

CombineTypedFilters Logical Optimization

CombineTypedFilters is a base logical optimization that combines two back to back (typed) filters into one that ultimately ends up as a single method call.

CombineTypedFilters is part of the Object Expressions Optimization fixed-point batch in the standard batches of the Catalyst Optimizer.

CombineTypedFilters is simply a Catalyst rule for transforming logical plans, i.e. Rule[LogicalPlan].



scala> :type spark
org.apache.spark.sql.SparkSession

// A query with two consecutive typed filters
val q = spark.range(10).filter(_ % 2 == 0).filter(_ == 0)
scala> q.queryExecution.optimizedPlan
...
TRACE SparkOptimizer:
=== Applying Rule org.apache.spark.sql.catalyst.optimizer.CombineTypedFilters ===
 TypedFilter <function1>, class java.lang.Long, [StructField(value,LongType,true)], newInstance(class java.lang.Long)      TypedFilter <function1>, class java.lang.Long, [StructField(value,LongType,true)], newInstance(class java.lang.Long)
!+- TypedFilter <function1>, class java.lang.Long, [StructField(value,LongType,true)], newInstance(class java.lang.Long)   +- Range (0, 10, step=1, splits=Some(8))
!   +- Range (0, 10, step=1, splits=Some(8))

TRACE SparkOptimizer: Fixed point reached for batch Typed Filter Optimization after 2 iterations.
DEBUG SparkOptimizer:
=== Result of Batch Typed Filter Optimization ===
 TypedFilter <function1>, class java.lang.Long, [StructField(value,LongType,true)], newInstance(class java.lang.Long)      TypedFilter <function1>, class java.lang.Long, [StructField(value,LongType,true)], newInstance(class java.lang.Long)
!+- TypedFilter <function1>, class java.lang.Long, [StructField(value,LongType,true)], newInstance(class java.lang.Long)   +- Range (0, 10, step=1, splits=Some(8))
!   +- Range (0, 10, step=1, splits=Some(8))
...

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

scala> :type spark

org.apache.spark.sql.SparkSession

// A query with two consecutive typed filters

val q = spark.range(10).filter(_ % 2 == 0).filter(_ == 0)

scala> q.queryExecution.optimizedPlan

...

TRACE SparkOptimizer:

=== Applying Rule org.apache.spark.sql.catalyst.optimizer.CombineTypedFilters ===

TypedFilter <function1>, class java.lang.Long, [StructField(value,LongType,true)], newInstance(class java.lang.Long) TypedFilter <function1>, class java.lang.Long, [StructField(value,LongType,true)], newInstance(class java.lang.Long)

!+- TypedFilter <function1>, class java.lang.Long, [StructField(value,LongType,true)], newInstance(class java.lang.Long) +- Range (0, 10, step=1, splits=Some(8))

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

TRACE SparkOptimizer: Fixed point reached for batch Typed Filter Optimization after 2 iterations.

DEBUG SparkOptimizer:

=== Result of Batch Typed Filter Optimization ===

TypedFilter <function1>, class java.lang.Long, [StructField(value,LongType,true)], newInstance(class java.lang.Long) TypedFilter <function1>, class java.lang.Long, [StructField(value,LongType,true)], newInstance(class java.lang.Long)

!+- TypedFilter <function1>, class java.lang.Long, [StructField(value,LongType,true)], newInstance(class java.lang.Long) +- Range (0, 10, step=1, splits=Some(8))

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

...

Executing Rule — `apply` Method



apply(plan: LogicalPlan): LogicalPlan

1

2

3

4

5

apply(plan: LogicalPlan): LogicalPlan

Note	`apply` is part of the Rule Contract to execute (apply) a rule on a TreeNode (e.g. LogicalPlan).

apply…FIXME

spark-sql 第14页

GetCurrentDatabase Logical Optimization

Executing Rule — apply Method

EliminateView Logical Optimization

Executing Rule — apply Method

EliminateSubqueryAliases Logical Optimization

Executing Rule — apply Method

EliminateSerialization Logical Optimization

Example — map followed by filter Logical Plan

Example — map followed by another map Logical Plan

Example — groupByKey followed by agg Logical Plan

Executing Rule — apply Method

DecimalAggregates Logical Optimization

Example: sum Aggregate Function on Decimal with Precision Smaller Than 9

Example: avg Aggregate Function on Decimal with Precision Smaller Than 12

Executing Rule — apply Method

CostBasedJoinReorder Logical Optimization — Join Reordering in Cost-Based Optimization

Executing Rule — apply Method

Reordering Logical Plan with Join Operators — reorder Internal Method

replaceWithOrderedJoin Internal Method

Extracting Consecutive Join Operators — extractInnerJoins Internal Method

ConstantFolding Logical Optimization

Executing Rule — apply Method

ComputeCurrentTime Logical Optimization

Executing Rule — apply Method

CombineUnions Logical Optimization

Executing Rule — apply Method

CombineTypedFilters Logical Optimization

Executing Rule — apply Method

欢迎关注：spark技术分享

关注公众号：spark技术分享

QQ咨询

回顶部

Executing Rule — `apply` Method

Executing Rule — `apply` Method

Executing Rule — `apply` Method

Example — `map` followed by `filter` Logical Plan

Example — `map` followed by another `map` Logical Plan

Example — `groupByKey` followed by `agg` Logical Plan

Executing Rule — `apply` Method

Executing Rule — `apply` Method

Executing Rule — `apply` Method

Reordering Logical Plan with Join Operators — `reorder` Internal Method

`replaceWithOrderedJoin` Internal Method

Extracting Consecutive Join Operators — `extractInnerJoins` Internal Method

Executing Rule — `apply` Method

Executing Rule — `apply` Method

Executing Rule — `apply` Method

Executing Rule — `apply` Method