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

RewriteCorrelatedScalarSubquery

2013-04-02admin阅读(1527)

RewriteCorrelatedScalarSubquery Logical Optimization

RewriteCorrelatedScalarSubquery is a base logical optimization that transforms logical plans with the following operators:

FIXME

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

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



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

// FIXME
// Demo: Filter + Aggregate
// Demo: Filter + UnaryNode

val plan = ???
val optimizedPlan = RewriteCorrelatedScalarSubquery(plan)

1

2

3

4

5

6

7

8

9

10

11

12

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

// FIXME

// Demo: Filter + Aggregate

// Demo: Filter + UnaryNode

val plan = ???

val optimizedPlan = RewriteCorrelatedScalarSubquery(plan)

`evalExpr` Internal Method



evalExpr(expr: Expression, bindings: Map[ExprId, Option[Any]]) : Option[Any]

1

2

3

4

5

evalExpr(expr: Expression, bindings: Map[ExprId, Option[Any]]) : Option[Any]

evalExpr…FIXME

Note	`evalExpr` is used exclusively when `RewriteCorrelatedScalarSubquery` is…FIXME

`evalAggOnZeroTups` Internal Method



evalAggOnZeroTups(expr: Expression) : Option[Any]

1

2

3

4

5

evalAggOnZeroTups(expr: Expression) : Option[Any]

evalAggOnZeroTups…FIXME

Note	`evalAggOnZeroTups` is used exclusively when `RewriteCorrelatedScalarSubquery` is…FIXME

`evalSubqueryOnZeroTups` Internal Method



evalSubqueryOnZeroTups(plan: LogicalPlan) : Option[Any]

1

2

3

4

5

evalSubqueryOnZeroTups(plan: LogicalPlan) : Option[Any]

evalSubqueryOnZeroTups…FIXME

Note	`evalSubqueryOnZeroTups` is used exclusively when `RewriteCorrelatedScalarSubquery` is requsted to constructLeftJoins.

`constructLeftJoins` Internal Method



constructLeftJoins(
  child: LogicalPlan,
  subqueries: ArrayBuffer[ScalarSubquery]): LogicalPlan

1

2

3

4

5

6

7

constructLeftJoins(

child: LogicalPlan,

subqueries: ArrayBuffer[ScalarSubquery]): LogicalPlan

constructLeftJoins…FIXME

Note	`constructLeftJoins` is used exclusively when `RewriteCorrelatedScalarSubquery` logical optimization is executed (i.e. applied to Aggregate, Project or Filter logical operators with correlated scalar subqueries)

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 transforms the input logical plan as follows:

For Aggregate operators, apply…FIXME
For Project operators, apply…FIXME
For Filter operators, apply…FIXME

Extracting ScalarSubquery Expressions with Children — `extractCorrelatedScalarSubqueries` Internal Method



extractCorrelatedScalarSubqueries[E <: Expression](
  expression: E,
  subqueries: ArrayBuffer[ScalarSubquery]): E

1

2

3

4

5

6

7

extractCorrelatedScalarSubqueries[E <: Expression](

expression: E,

subqueries: ArrayBuffer[ScalarSubquery]): E

extractCorrelatedScalarSubqueries finds all ScalarSubquery expressions with at least one child in the input expression and adds them to the input subqueries collection.

extractCorrelatedScalarSubqueries traverses the input expression down (the expression tree) and, every time a ScalarSubquery with at least one child is found, returns the head of the output attributes of the subquery plan.

In the end, extractCorrelatedScalarSubqueries returns the rewritten expression.

Note	`extractCorrelatedScalarSubqueries` uses scala.collection.mutable.ArrayBuffer and mutates an instance inside (i.e. adds `ScalarSubquery` expressions) that makes for two output values, i.e. the rewritten expression and the `ScalarSubquery` expressions.

Note	`extractCorrelatedScalarSubqueries` is used exclusively when `RewriteCorrelatedScalarSubquery` is executed (i.e. applied to a logical plan).

ReplaceExpressions

2013-04-01admin阅读(1388)

ReplaceExpressions Logical Optimization

ReplaceExpressions is a base logical optimization that replaces RuntimeReplaceable expressions with their single child expression.

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

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



val query = sql("select ifnull(NULL, array('2')) from values 1")
val analyzedPlan = query.queryExecution.analyzed
scala> println(analyzedPlan.numberedTreeString)
00 Project [ifnull(null, array(2)) AS ifnull(NULL, array('2'))#3]
01 +- LocalRelation [col1#2]

import org.apache.spark.sql.catalyst.optimizer.ReplaceExpressions
val optimizedPlan = ReplaceExpressions(analyzedPlan)
scala> println(optimizedPlan.numberedTreeString)
00 Project [coalesce(cast(null as array<string>), cast(array(2) as array<string>)) AS ifnull(NULL, array('2'))#3]
01 +- LocalRelation [col1#2]

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

val query = sql("select ifnull(NULL, array('2')) from values 1")

val analyzedPlan = query.queryExecution.analyzed

scala> println(analyzedPlan.numberedTreeString)

00 Project [ifnull(null, array(2)) AS ifnull(NULL, array('2'))#3]

01 +- LocalRelation [col1#2]

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

val optimizedPlan = ReplaceExpressions(analyzedPlan)

scala> println(optimizedPlan.numberedTreeString)

00 Project [coalesce(cast(null as array<string>), cast(array(2) as array<string>)) AS ifnull(NULL, array('2'))#3]

01 +- LocalRelation [col1#2]

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 all Catalyst expressions (in the input LogicalPlan) and replaces a RuntimeReplaceable expression into its single child.

PushPredicateThroughJoin

2013-03-31admin阅读(1355)

PushPredicateThroughJoin Logical Optimization

PushPredicateThroughJoin is a base logical optimization that FIXME.

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

PushPredicateThroughJoin 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

PullupCorrelatedPredicates

2013-03-30admin阅读(1679)

PullupCorrelatedPredicates Logical Optimization

PullupCorrelatedPredicates is a base logical optimization that transforms logical plans with the following operators:

Filter operators with an Aggregate child operator
UnaryNode operators

PullupCorrelatedPredicates is part of the Pullup Correlated Expressions once-executed batch in the standard batches of the Catalyst Optimizer.

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



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

// FIXME
// Demo: Filter + Aggregate
// Demo: Filter + UnaryNode

val plan = ???
val optimizedPlan = PullupCorrelatedPredicates(plan)

1

2

3

4

5

6

7

8

9

10

11

12

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

// FIXME

// Demo: Filter + Aggregate

// Demo: Filter + UnaryNode

val plan = ???

val optimizedPlan = PullupCorrelatedPredicates(plan)

PullupCorrelatedPredicates uses PredicateHelper for…FIXME

`pullOutCorrelatedPredicates` Internal Method



pullOutCorrelatedPredicates(
  sub: LogicalPlan,
  outer: Seq[LogicalPlan]): (LogicalPlan, Seq[Expression])

1

2

3

4

5

6

7

pullOutCorrelatedPredicates(

sub: LogicalPlan,

outer: Seq[LogicalPlan]): (LogicalPlan, Seq[Expression])

pullOutCorrelatedPredicates…FIXME

Note	`pullOutCorrelatedPredicates` is used exclusively when `PullupCorrelatedPredicates` is requested to rewriteSubQueries.

`rewriteSubQueries` Internal Method



rewriteSubQueries(plan: LogicalPlan, outerPlans: Seq[LogicalPlan]): LogicalPlan

1

2

3

4

5

rewriteSubQueries(plan: LogicalPlan, outerPlans: Seq[LogicalPlan]): LogicalPlan

rewriteSubQueries…FIXME

Note	`rewriteSubQueries` is used exclusively when `PullupCorrelatedPredicates` is executed (i.e. applied to a logical plan).

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 transforms the input logical plan as follows:

For Filter operators with an Aggregate child operator, apply rewriteSubQueries with the Filter and the Aggregate and its child as the outer plans
For UnaryNode operators, apply rewriteSubQueries with the operator and its children as the outer plans

PropagateEmptyRelation

2013-03-29admin阅读(1607)

PropagateEmptyRelation Logical Optimization

PropagateEmptyRelation is a base logical optimization that collapses plans with empty LocalRelation logical operators, e.g. explode or join.

PropagateEmptyRelation is part of the LocalRelation fixed-point batch in the standard batches of the Catalyst Optimizer.

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

Explode



scala> val emp = spark.emptyDataset[Seq[String]]
emp: org.apache.spark.sql.Dataset[Seq[String]] = [value: array<string>]

scala> emp.select(explode($"value")).show
+---+
|col|
+---+
+---+

scala> emp.select(explode($"value")).explain(true)
== Parsed Logical Plan ==
'Project [explode('value) AS List()]
+- LocalRelation <empty>, [value#77]

== Analyzed Logical Plan ==
col: string
Project [col#89]
+- Generate explode(value#77), false, false, [col#89]
   +- LocalRelation <empty>, [value#77]

== Optimized Logical Plan ==
LocalRelation <empty>, [col#89]

== Physical Plan ==
LocalTableScan <empty>, [col#89]

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

scala> val emp = spark.emptyDataset[Seq[String]]

emp: org.apache.spark.sql.Dataset[Seq[String]] = [value: array<string>]

scala> emp.select(explode($"value")).show

+---+

|col|

+---+

scala> emp.select(explode($"value")).explain(true)

== Parsed Logical Plan ==

'Project [explode('value) AS List()]

+- LocalRelation <empty>, [value#77]

== Analyzed Logical Plan ==

col: string

Project [col#89]

+- Generate explode(value#77), false, false, [col#89]

+- LocalRelation <empty>, [value#77]

== Optimized Logical Plan ==

LocalRelation <empty>, [col#89]

== Physical Plan ==

LocalTableScan <empty>, [col#89]

Join



scala> spark.emptyDataset[Int].join(spark.range(1)).explain(extended = true)
...
TRACE SparkOptimizer:
=== Applying Rule org.apache.spark.sql.catalyst.optimizer.PropagateEmptyRelation ===
!Join Inner                                LocalRelation <empty>, [value#40, id#42L]
!:- LocalRelation <empty>, [value#40]
!+- Range (0, 1, step=1, splits=Some(8))

TRACE SparkOptimizer: Fixed point reached for batch LocalRelation after 2 iterations.
DEBUG SparkOptimizer:
=== Result of Batch LocalRelation ===
!Join Inner                                LocalRelation <empty>, [value#40, id#42L]
!:- LocalRelation <empty>, [value#40]
!+- Range (0, 1, step=1, splits=Some(8))
...
== Parsed Logical Plan ==
Join Inner
:- LocalRelation <empty>, [value#40]
+- Range (0, 1, step=1, splits=Some(8))

== Analyzed Logical Plan ==
value: int, id: bigint
Join Inner
:- LocalRelation <empty>, [value#40]
+- Range (0, 1, step=1, splits=Some(8))

== Optimized Logical Plan ==
LocalRelation <empty>, [value#40, id#42L]

== Physical Plan ==
LocalTableScan <empty>, [value#40, id#42L]

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

scala> spark.emptyDataset[Int].join(spark.range(1)).explain(extended = true)

...

TRACE SparkOptimizer:

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

!Join Inner LocalRelation <empty>, [value#40, id#42L]

!:- LocalRelation <empty>, [value#40]

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

TRACE SparkOptimizer: Fixed point reached for batch LocalRelation after 2 iterations.

DEBUG SparkOptimizer:

=== Result of Batch LocalRelation ===

!Join Inner LocalRelation <empty>, [value#40, id#42L]

!:- LocalRelation <empty>, [value#40]

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

...

== Parsed Logical Plan ==

Join Inner

:- LocalRelation <empty>, [value#40]

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

== Analyzed Logical Plan ==

value: int, id: bigint

Join Inner

:- LocalRelation <empty>, [value#40]

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

== Optimized Logical Plan ==

LocalRelation <empty>, [value#40, id#42L]

== Physical Plan ==

LocalTableScan <empty>, [value#40, id#42L]

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

OptimizeSubqueries

2013-03-28admin阅读(1619)

OptimizeSubqueries Logical Optimization

OptimizeSubqueries is a base logical optimization that FIXME.

OptimizeSubqueries is part of the Subquery once-executed batch in the standard batches of the Catalyst Optimizer.

OptimizeSubqueries 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

使用 json定义spark sql schema 代码例子

2013-03-27admin阅读(2526)

package streaming.core

import org.apache.spark.sql.SparkSession
import org.scalatest.{BeforeAndAfter, BeforeAndAfterAll, FunSuite}
import org.scalatest.concurrent.Eventually
import streaming.util.Logging
import org.apache.spark.sql.types._
import org.apache.spark.sql.{streaming, functions => F, _}


/**
  * Created by sunbiaobiao on 2019/3/27.
  */
class SchemaTest extends FunSuite
  with BeforeAndAfter
  with BeforeAndAfterAll
  with Eventually  with Logging {

  override def beforeAll {}

  test("spark sql schema") {

    val spark = SparkSession.builder().getOrCreate()

    import  spark.implicits._
    import  spark._

    val addressesSchema = new StructType().add($"city".string).add($"state".string).add($"zip".string)

    val schema = new StructType().add($"firstName".string).add($"lastName".string).add($"email".string).add($"addresses".array(addressesSchema)).add("sons", ArrayType(MapType(StringType, new StructType().add($"firstName".string))))add("keys", MapType(StringType, StringType))

    println(schema.prettyJson)

    import org.apache.spark.sql.types.DataType

    //val schemaAsJson = schema.prettyJson
    // 打印出来的json 如下
    val schemaAsJson =
      """
        |{
        |  "type" : "struct",
        |  "fields" : [ {
        |    "name" : "firstName",
        |    "type" : "string",
        |    "nullable" : true,
        |    "metadata" : { }
        |  }, {
        |    "name" : "lastName",
        |    "type" : "string",
        |    "nullable" : true,
        |    "metadata" : { }
        |  }, {
        |    "name" : "email",
        |    "type" : "string",
        |    "nullable" : true,
        |    "metadata" : { }
        |  }, {
        |    "name" : "addresses",
        |    "type" : {
        |      "type" : "array",
        |      "elementType" : {
        |        "type" : "struct",
        |        "fields" : [ {
        |          "name" : "city",
        |          "type" : "string",
        |          "nullable" : true,
        |          "metadata" : { }
        |        }, {
        |          "name" : "state",
        |          "type" : "string",
        |          "nullable" : true,
        |          "metadata" : { }
        |        }, {
        |          "name" : "zip",
        |          "type" : "string",
        |          "nullable" : true,
        |          "metadata" : { }
        |        } ]
        |      },
        |      "containsNull" : true
        |    },
        |    "nullable" : true,
        |    "metadata" : { }
        |  }, {
        |    "name" : "sons",
        |    "type" : {
        |      "type" : "array",
        |      "elementType" : {
        |        "type" : "map",
        |        "keyType" : "string",
        |        "valueType" : {
        |          "type" : "struct",
        |          "fields" : [ {
        |            "name" : "firstName",
        |            "type" : "string",
        |            "nullable" : true,
        |            "metadata" : { }
        |          } ]
        |        },
        |        "valueContainsNull" : true
        |      },
        |      "containsNull" : true
        |    },
        |    "nullable" : true,
        |    "metadata" : { }
        |  }, {
        |    "name" : "keys",
        |    "type" : {
        |      "type" : "map",
        |      "keyType" : "string",
        |      "valueType" : "string",
        |      "valueContainsNull" : true
        |    },
        |    "nullable" : true,
        |    "metadata" : { }
        |  } ]
        |}
      """.stripMargin

    val dt = DataType.fromJson(schemaAsJson)


    val rawJsons = Seq("""
  {
    "firstName" : "Jacek",
    "lastName" : "Laskowski",
    "email" : "jacek@japila.pl",
    "addresses" : [
      {
        "city" : "Warsaw",
        "state" : "N/A",
        "zip" : "02-791"
      }
    ]
  }
""").toDF("rawjson")

    val people = rawJsons
      .select(F.from_json($"rawjson", dt, Map.empty[String, String]) as "json")
      .select("json.*") // <-- flatten the struct field
      .withColumn("address", F.explode($"addresses")) // <-- explode the array field
      .drop("addresses")  // <-- no longer needed
      .select("firstName", "lastName", "email", "address.*") // <-- flatten the struct field
  }

}

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

96

97

98

99

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

package streaming.core

import org.apache.spark.sql.SparkSession

import org.scalatest.{BeforeAndAfter, BeforeAndAfterAll, FunSuite}

import org.scalatest.concurrent.Eventually

import streaming.util.Logging

import org.apache.spark.sql.types._

import org.apache.spark.sql.{streaming, functions => F, _}

/**

* Created by sunbiaobiao on 2019/3/27.

*/

class SchemaTest extends FunSuite

with BeforeAndAfter

with BeforeAndAfterAll

with Eventually with Logging {

override def beforeAll {}

test("spark sql schema") {

val spark = SparkSession.builder().getOrCreate()

import spark.implicits._

import spark._

val addressesSchema = new StructType().add($"city".string).add($"state".string).add($"zip".string)

val schema = new StructType().add($"firstName".string).add($"lastName".string).add($"email".string).add($"addresses".array(addressesSchema)).add("sons", ArrayType(MapType(StringType, new StructType().add($"firstName".string))))add("keys", MapType(StringType, StringType))

println(schema.prettyJson)

import org.apache.spark.sql.types.DataType

//val schemaAsJson = schema.prettyJson

// 打印出来的json 如下

val schemaAsJson =

"""

|{

| "type" : "struct",

| "fields" : [ {

| "name" : "firstName",

| "type" : "string",

| "nullable" : true,

| "metadata" : { }

| }, {

| "name" : "lastName",

| "type" : "string",

| "nullable" : true,

| "metadata" : { }

| }, {

| "name" : "email",

| "type" : "string",

| "nullable" : true,

| "metadata" : { }

| }, {

| "name" : "addresses",

| "type" : {

| "type" : "array",

| "elementType" : {

| "type" : "struct",

| "fields" : [ {

| "name" : "city",

| "type" : "string",

| "nullable" : true,

| "metadata" : { }

| }, {

| "name" : "state",

| "type" : "string",

| "nullable" : true,

| "metadata" : { }

| }, {

| "name" : "zip",

| "type" : "string",

| "nullable" : true,

| "metadata" : { }

| } ]

| },

| "containsNull" : true

| },

| "nullable" : true,

| "metadata" : { }

| }, {

| "name" : "sons",

| "type" : {

| "type" : "array",

| "elementType" : {

| "type" : "map",

| "keyType" : "string",

| "valueType" : {

| "type" : "struct",

| "fields" : [ {

| "name" : "firstName",

| "type" : "string",

| "nullable" : true,

| "metadata" : { }

| } ]

| },

| "valueContainsNull" : true

| },

| "containsNull" : true

| },

| "nullable" : true,

| "metadata" : { }

| }, {

| "name" : "keys",

| "type" : {

| "type" : "map",

| "keyType" : "string",

| "valueType" : "string",

| "valueContainsNull" : true

| },

| "nullable" : true,

| "metadata" : { }

| } ]

|}

""".stripMargin

val dt = DataType.fromJson(schemaAsJson)

val rawJsons = Seq("""

{

"firstName" : "Jacek",

"lastName" : "Laskowski",

"email" : "jacek@japila.pl",

"addresses" : [

{

"city" : "Warsaw",

"state" : "N/A",

"zip" : "02-791"

}

]

}

""").toDF("rawjson")

val people = rawJsons

.select(F.from_json($"rawjson", dt, Map.empty[String, String]) as "json")

.select("json.*") // <-- flatten the struct field

.withColumn("address", F.explode($"addresses")) // <-- explode the array field

.drop("addresses") // <-- no longer needed

.select("firstName", "lastName", "email", "address.*") // <-- flatten the struct field

}

OptimizeIn

2013-03-27admin阅读(1483)

OptimizeIn Logical Optimization

OptimizeIn is a base logical optimization that transforms logical plans with In predicate expressions as follows:

Replaces an In expression that has an empty list and the value expression not nullable to false
Eliminates duplicates of Literal expressions in an In predicate expression that is inSetConvertible
Replaces an In predicate expression that is inSetConvertible with InSet expressions when the number of literal expressions in the list expression is greater than spark.sql.optimizer.inSetConversionThreshold internal configuration property (default: 10)

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

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



// Use Catalyst DSL to define a logical plan

// HACK: Disable symbolToColumn implicit conversion
// It is imported automatically in spark-shell (and makes demos impossible)
// implicit def symbolToColumn(s: Symbol): org.apache.spark.sql.ColumnName
trait ThatWasABadIdea
implicit def symbolToColumn(ack: ThatWasABadIdea) = ack

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

import org.apache.spark.sql.catalyst.plans.logical.LocalRelation
val rel = LocalRelation('a.int, 'b.int, 'c.int)

import org.apache.spark.sql.catalyst.expressions.{In, Literal}
val plan = rel
  .where(In('a, Seq[Literal](1, 2, 3)))
  .analyze
scala> println(plan.numberedTreeString)
00 Filter a#6 IN (1,2,3)
01 +- LocalRelation <empty>, [a#6, b#7, c#8]

// In --> InSet
spark.conf.set("spark.sql.optimizer.inSetConversionThreshold", 0)

import org.apache.spark.sql.catalyst.optimizer.OptimizeIn
val optimizedPlan = OptimizeIn(plan)
scala> println(optimizedPlan.numberedTreeString)
00 Filter a#6 INSET (1,2,3)
01 +- LocalRelation <empty>, [a#6, b#7, c#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

// Use Catalyst DSL to define a logical plan

// HACK: Disable symbolToColumn implicit conversion

// It is imported automatically in spark-shell (and makes demos impossible)

// implicit def symbolToColumn(s: Symbol): org.apache.spark.sql.ColumnName

trait ThatWasABadIdea

implicit def symbolToColumn(ack: ThatWasABadIdea) = ack

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

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

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

val rel = LocalRelation('a.int, 'b.int, 'c.int)

import org.apache.spark.sql.catalyst.expressions.{In, Literal}

val plan = rel

.where(In('a, Seq[Literal](1, 2, 3)))

.analyze

scala> println(plan.numberedTreeString)

00 Filter a#6 IN (1,2,3)

01 +- LocalRelation <empty>, [a#6, b#7, c#8]

// In --> InSet

spark.conf.set("spark.sql.optimizer.inSetConversionThreshold", 0)

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

val optimizedPlan = OptimizeIn(plan)

scala> println(optimizedPlan.numberedTreeString)

00 Filter a#6 INSET (1,2,3)

01 +- LocalRelation <empty>, [a#6, b#7, c#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

NullPropagation

2013-03-26admin阅读(1523)

NullPropagation Logical Optimization — Nullability (NULL Value) Propagation

NullPropagation is a base logical optimization that FIXME.

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

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

Example: Count Aggregate Operator with Nullable Expressions Only

NullPropagation optimization rewrites Count aggregate expressions that include expressions that are all nullable to Cast(Literal(0L)).



val table = (0 to 9).toDF("num").as[Int]

// NullPropagation applied
scala> table.select(countDistinct($"num" === null)).explain(true)
== Parsed Logical Plan ==
'Project [count(distinct ('num = null)) AS count(DISTINCT (num = NULL))#45]
+- Project [value#1 AS num#3]
   +- LocalRelation [value#1]

== Analyzed Logical Plan ==
count(DISTINCT (num = NULL)): bigint
Aggregate [count(distinct (num#3 = cast(null as int))) AS count(DISTINCT (num = NULL))#45L]
+- Project [value#1 AS num#3]
   +- LocalRelation [value#1]

== Optimized Logical Plan ==
Aggregate [0 AS count(DISTINCT (num = NULL))#45L] // <-- HERE
+- LocalRelation

== Physical Plan ==
*HashAggregate(keys=[], functions=[], output=[count(DISTINCT (num = NULL))#45L])
+- Exchange SinglePartition
   +- *HashAggregate(keys=[], functions=[], output=[])
      +- LocalTableScan

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

val table = (0 to 9).toDF("num").as[Int]

// NullPropagation applied

scala> table.select(countDistinct($"num" === null)).explain(true)

== Parsed Logical Plan ==

'Project [count(distinct ('num = null)) AS count(DISTINCT (num = NULL))#45]

+- Project [value#1 AS num#3]

+- LocalRelation [value#1]

== Analyzed Logical Plan ==

count(DISTINCT (num = NULL)): bigint

Aggregate [count(distinct (num#3 = cast(null as int))) AS count(DISTINCT (num = NULL))#45L]

+- Project [value#1 AS num#3]

+- LocalRelation [value#1]

== Optimized Logical Plan ==

Aggregate [0 AS count(DISTINCT (num = NULL))#45L] // <-- HERE

+- LocalRelation

== Physical Plan ==

*HashAggregate(keys=[], functions=[], output=[count(DISTINCT (num = NULL))#45L])

+- Exchange SinglePartition

+- *HashAggregate(keys=[], functions=[], output=[])

+- LocalTableScan

Example: Count Aggregate Operator with Non-Nullable Non-Distinct Expressions

NullPropagation optimization rewrites any non-nullable non-distinct Count aggregate expressions to Literal(1).



val table = (0 to 9).toDF("num").as[Int]

// NullPropagation applied
// current_timestamp() is a non-nullable expression (see the note below)
val query = table.select(count(current_timestamp()) as "count")

scala> println(query.queryExecution.optimizedPlan)
Aggregate [count(1) AS count#64L]
+- LocalRelation

// NullPropagation skipped
val tokens = Seq((0, null), (1, "hello")).toDF("id", "word")
val query = tokens.select(count("word") as "count")

scala> println(query.queryExecution.optimizedPlan)
Aggregate [count(word#55) AS count#71L]
+- LocalRelation [word#55]

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

val table = (0 to 9).toDF("num").as[Int]

// NullPropagation applied

// current_timestamp() is a non-nullable expression (see the note below)

val query = table.select(count(current_timestamp()) as "count")

scala> println(query.queryExecution.optimizedPlan)

Aggregate [count(1) AS count#64L]

+- LocalRelation

// NullPropagation skipped

val tokens = Seq((0, null), (1, "hello")).toDF("id", "word")

val query = tokens.select(count("word") as "count")

scala> println(query.queryExecution.optimizedPlan)

Aggregate [count(word#55) AS count#71L]

+- LocalRelation [word#55]

Note

Count aggregate expression represents count function internally.



import org.apache.spark.sql.catalyst.expressions.aggregate.Count
import org.apache.spark.sql.functions.count

scala> count("*").expr.children(0).asInstanceOf[Count]
res0: org.apache.spark.sql.catalyst.expressions.aggregate.Count = count(1)

1

2

3

4

5

6

7

8

9

import org.apache.spark.sql.catalyst.expressions.aggregate.Count

import org.apache.spark.sql.functions.count

scala> count("*").expr.children(0).asInstanceOf[Count]

res0: org.apache.spark.sql.catalyst.expressions.aggregate.Count = count(1)

Note

current_timestamp() function is non-nullable expression.



import org.apache.spark.sql.catalyst.expressions.CurrentTimestamp
import org.apache.spark.sql.functions.current_timestamp

scala> current_timestamp().expr.asInstanceOf[CurrentTimestamp].nullable
res38: Boolean = false

1

2

3

4

5

6

7

8

9

import org.apache.spark.sql.catalyst.expressions.CurrentTimestamp

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

scala> current_timestamp().expr.asInstanceOf[CurrentTimestamp].nullable

res38: Boolean = false

Example



val table = (0 to 9).toDF("num").as[Int]
val query = table.where('num === null)

scala> query.explain(extended = true)
== Parsed Logical Plan ==
'Filter ('num = null)
+- Project [value#1 AS num#3]
   +- LocalRelation [value#1]

== Analyzed Logical Plan ==
num: int
Filter (num#3 = cast(null as int))
+- Project [value#1 AS num#3]
   +- LocalRelation [value#1]

== Optimized Logical Plan ==
LocalRelation <empty>, [num#3]

== Physical Plan ==
LocalTableScan <empty>, [num#3]

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

val table = (0 to 9).toDF("num").as[Int]

val query = table.where('num === null)

scala> query.explain(extended = true)

== Parsed Logical Plan ==

'Filter ('num = null)

+- Project [value#1 AS num#3]

+- LocalRelation [value#1]

== Analyzed Logical Plan ==

num: int

Filter (num#3 = cast(null as int))

+- Project [value#1 AS num#3]

+- LocalRelation [value#1]

== Optimized Logical Plan ==

LocalRelation <empty>, [num#3]

== Physical Plan ==

LocalTableScan <empty>, [num#3]

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

LimitPushDown

2013-03-25admin阅读(1720)

LimitPushDown Logical Optimization

LimitPushDown is a base logical optimization that transforms the following logical plans:

LocalLimit with Union
LocalLimit with Join

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

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



// test datasets
scala> val ds1 = spark.range(4)
ds1: org.apache.spark.sql.Dataset[Long] = [value: bigint]

scala> val ds2 = spark.range(2)
ds2: org.apache.spark.sql.Dataset[Long] = [value: bigint]

// Case 1. Rather than `LocalLimit` of `Union` do `Union` of `LocalLimit`
scala> ds1.union(ds2).limit(2).explain(true)
== Parsed Logical Plan ==
GlobalLimit 2
+- LocalLimit 2
   +- Union
      :- Range (0, 4, step=1, splits=Some(8))
      +- Range (0, 2, step=1, splits=Some(8))

== Analyzed Logical Plan ==
id: bigint
GlobalLimit 2
+- LocalLimit 2
   +- Union
      :- Range (0, 4, step=1, splits=Some(8))
      +- Range (0, 2, step=1, splits=Some(8))

== Optimized Logical Plan ==
GlobalLimit 2
+- LocalLimit 2
   +- Union
      :- LocalLimit 2
      :  +- Range (0, 4, step=1, splits=Some(8))
      +- LocalLimit 2
         +- Range (0, 2, step=1, splits=Some(8))

== Physical Plan ==
CollectLimit 2
+- Union
   :- *LocalLimit 2
   :  +- *Range (0, 4, step=1, splits=Some(8))
   +- *LocalLimit 2
      +- *Range (0, 2, 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

// test datasets

scala> val ds1 = spark.range(4)

ds1: org.apache.spark.sql.Dataset[Long] = [value: bigint]

scala> val ds2 = spark.range(2)

ds2: org.apache.spark.sql.Dataset[Long] = [value: bigint]

// Case 1. Rather than `LocalLimit` of `Union` do `Union` of `LocalLimit`

scala> ds1.union(ds2).limit(2).explain(true)

== Parsed Logical Plan ==

GlobalLimit 2

+- LocalLimit 2

+- Union

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

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

== Analyzed Logical Plan ==

id: bigint

GlobalLimit 2

+- LocalLimit 2

+- Union

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

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

== Optimized Logical Plan ==

GlobalLimit 2

+- LocalLimit 2

+- Union

:- LocalLimit 2

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

+- LocalLimit 2

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

== Physical Plan ==

CollectLimit 2

+- Union

:- *LocalLimit 2

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

+- *LocalLimit 2

+- *Range (0, 2, 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

Creating LimitPushDown Instance

LimitPushDown takes the following when created:

CatalystConf

LimitPushDown initializes the internal registries and counters.

Note	`LimitPushDown` is created when

spark-sql 第13页

RewriteCorrelatedScalarSubquery Logical Optimization

evalExpr Internal Method

evalAggOnZeroTups Internal Method

evalSubqueryOnZeroTups Internal Method

constructLeftJoins Internal Method

Executing Rule — apply Method

Extracting ScalarSubquery Expressions with Children — extractCorrelatedScalarSubqueries Internal Method

ReplaceExpressions Logical Optimization

Executing Rule — apply Method

PushPredicateThroughJoin Logical Optimization

Executing Rule — apply Method

PullupCorrelatedPredicates Logical Optimization

pullOutCorrelatedPredicates Internal Method

rewriteSubQueries Internal Method

Executing Rule — apply Method

PropagateEmptyRelation Logical Optimization

Explode

Join

Executing Rule — apply Method

OptimizeSubqueries Logical Optimization

Executing Rule — apply Method

OptimizeIn Logical Optimization

Executing Rule — apply Method

NullPropagation Logical Optimization — Nullability (NULL Value) Propagation

Example: Count Aggregate Operator with Nullable Expressions Only

Example: Count Aggregate Operator with Non-Nullable Non-Distinct Expressions

Example

Executing Rule — apply Method

LimitPushDown Logical Optimization

Executing Rule — apply Method

Creating LimitPushDown Instance

欢迎关注：spark技术分享

关注公众号：spark技术分享

QQ咨询

回顶部

`evalExpr` Internal Method

`evalAggOnZeroTups` Internal Method

`evalSubqueryOnZeroTups` Internal Method

`constructLeftJoins` Internal Method

Executing Rule — `apply` Method

Extracting ScalarSubquery Expressions with Children — `extractCorrelatedScalarSubqueries` Internal Method

Executing Rule — `apply` Method

Executing Rule — `apply` Method

`pullOutCorrelatedPredicates` Internal Method

`rewriteSubQueries` Internal Method

Executing Rule — `apply` Method

Executing Rule — `apply` Method

Executing Rule — `apply` Method

Executing Rule — `apply` Method

Executing Rule — `apply` Method

Executing Rule — `apply` Method