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

ResolvedStar

2012-09-15admin阅读(1091)

ResolvedStar

ResolvedStar is…FIXME

RankLike Contract

2012-09-14admin阅读(1092)

RankLike Contract

RankLike is…FIXME

PrettyAttribute

2012-09-13admin阅读(1102)

PrettyAttribute

PrettyAttribute is…FIXME

PlanExpression

2012-09-12admin阅读(1409)

PlanExpression Contract for Expressions with Query Plans

PlanExpression is the contract for Catalyst expressions that contain a QueryPlan.



package org.apache.spark.sql.catalyst.expressions

abstract class PlanExpression[T <: QueryPlan[_]] extends Expression {
  // only required methods that have no implementation
  // the others follow
  def exprId: ExprId
  def plan: T
  def withNewPlan(plan: T): PlanExpression[T]
}

1

2

3

4

5

6

7

8

9

10

11

12

13

package org.apache.spark.sql.catalyst.expressions

abstract class PlanExpression[T <: QueryPlan[_]] extends Expression {

// only required methods that have no implementation

// the others follow

def exprId: ExprId

def plan: T

def withNewPlan(plan: T): PlanExpression[T]

}

Table 1. PlanExpression Contract
Method	Description
`exprId`	Used when…FIXME
`plan`	Used when…FIXME
`withNewPlan`	Used when…FIXME

Table 2. PlanExpressions
PlanExpression	Description
ExecSubqueryExpression
SubqueryExpression

ParseToTimestamp

2012-09-11admin阅读(1323)

ParseToTimestamp Expression

ParseToTimestamp is a RuntimeReplaceable expression that represents the to_timestamp function (in logical query plans).



// DEMO to_timestamp(s: Column): Column
import java.sql.Timestamp
import java.time.LocalDateTime
val times = Seq(Timestamp.valueOf(LocalDateTime.of(2018, 5, 30, 0, 0, 0)).toString).toDF("time")
scala> times.printSchema
root
 |-- time: string (nullable = true)

import org.apache.spark.sql.functions.to_timestamp
val q = times.select(to_timestamp($"time") as "ts")
scala> q.printSchema
root
 |-- ts: timestamp (nullable = true)

val plan = q.queryExecution.analyzed
scala> println(plan.numberedTreeString)
00 Project [to_timestamp('time, None) AS ts#29]
01 +- Project [value#16 AS time#18]
02    +- LocalRelation [value#16]

import org.apache.spark.sql.catalyst.expressions.ParseToTimestamp
val ptt = plan.expressions.head.children.head.asInstanceOf[ParseToTimestamp]
scala> println(ptt.numberedTreeString)
00 to_timestamp('time, None)
01 +- cast(time#18 as timestamp)
02    +- time#18: string

// FIXME DEMO to_timestamp(s: Column, fmt: String): Column

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

// DEMO to_timestamp(s: Column): Column

import java.sql.Timestamp

import java.time.LocalDateTime

val times = Seq(Timestamp.valueOf(LocalDateTime.of(2018, 5, 30, 0, 0, 0)).toString).toDF("time")

scala> times.printSchema

root

|-- time: string (nullable = true)

import org.apache.spark.sql.functions.to_timestamp

val q = times.select(to_timestamp($"time") as "ts")

scala> q.printSchema

root

|-- ts: timestamp (nullable = true)

val plan = q.queryExecution.analyzed

scala> println(plan.numberedTreeString)

00 Project [to_timestamp('time, None) AS ts#29]

01 +- Project [value#16 AS time#18]

02 +- LocalRelation [value#16]

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

val ptt = plan.expressions.head.children.head.asInstanceOf[ParseToTimestamp]

scala> println(ptt.numberedTreeString)

00 to_timestamp('time, None)

01 +- cast(time#18 as timestamp)

02 +- time#18: string

// FIXME DEMO to_timestamp(s: Column, fmt: String): Column

As a RuntimeReplaceable expression, ParseToTimestamp is replaced by Catalyst Optimizer with the child expression:

Cast(left, TimestampType) for to_timestamp(s: Column): Column function
Cast(UnixTimestamp(left, format), TimestampType) for to_timestamp(s: Column, fmt: String): Column function



// FIXME DEMO Conversion to `Cast(left, TimestampType)`
// FIXME DEMO Conversion to `Cast(UnixTimestamp(left, format), TimestampType)`

1

2

3

4

5

6

// FIXME DEMO Conversion to `Cast(left, TimestampType)`

// FIXME DEMO Conversion to `Cast(UnixTimestamp(left, format), TimestampType)`

Creating ParseToTimestamp Instance

ParseToTimestamp takes the following when created:

Left expression
format expression
Child expression

ParseToDate

2012-09-10admin阅读(1349)

ParseToDate Expression

ParseToDate is a RuntimeReplaceable expression that represents the to_date function (in logical query plans).



// DEMO to_date(e: Column): Column
// DEMO to_date(e: Column, fmt: String): Column

1

2

3

4

5

6

// DEMO to_date(e: Column): Column

// DEMO to_date(e: Column, fmt: String): Column

As a RuntimeReplaceable expression, ParseToDate is replaced by Catalyst Optimizer with the child expression:

Cast(left, DateType) for to_date(e: Column): Column function
Cast(Cast(UnixTimestamp(left, format), TimestampType), DateType) for to_date(e: Column, fmt: String): Column function



// FIXME DEMO Conversion to `Cast(left, DateType)`
// FIXME DEMO Conversion to `Cast(Cast(UnixTimestamp(left, format), TimestampType), DateType)`

1

2

3

4

5

6

// FIXME DEMO Conversion to `Cast(left, DateType)`

// FIXME DEMO Conversion to `Cast(Cast(UnixTimestamp(left, format), TimestampType), DateType)`

Creating ParseToDate Instance

ParseToDate takes the following when created:

Left expression
format expression
Child expression

OffsetWindowFunction Contract — Unevaluable Window Function Expressions

2012-09-09admin阅读(1542)

OffsetWindowFunction Contract — Unevaluable Window Function Expressions

OffsetWindowFunction is the base of window function expressions that are unevaluable and ImplicitCastInputTypes.

Note	An unevaluable expression cannot be evaluated to produce a value (neither in interpreted nor code-generated expression evaluations) and has to be resolved (replaced) to some other expressions or logical operators at analysis or optimization phases or they fail analysis.



package org.apache.spark.sql.catalyst.expressions

abstract class OffsetWindowFunction ... {
  // only required properties (vals and methods) that have no implementation
  // the others follow
  val default: Expression
  val direction: SortDirection
  val input: Expression
  val offset: Expression
}

1

2

3

4

5

6

7

8

9

10

11

12

13

14

package org.apache.spark.sql.catalyst.expressions

abstract class OffsetWindowFunction ... {

// only required properties (vals and methods) that have no implementation

// the others follow

val default: Expression

val direction: SortDirection

val input: Expression

val offset: Expression

}

Table 1. (Subset of) OffsetWindowFunction Contract
Property	Description
`default`
`direction`
`input`
`offset`

OffsetWindowFunction uses the input, offset and default expressions as the children.

OffsetWindowFunction is not foldable.

OffsetWindowFunction is nullable when the default is not defined or the default or the input expressions are.

When requested for the dataType, OffsetWindowFunction simply requests the input expression for the data type.

When requested for the inputTypes, OffsetWindowFunction returns the AnyDataType, IntegerType with the data type of the input expression and the NullType.

OffsetWindowFunction uses the following text representation (i.e. toString):



[prettyName]([input], [offset], [default])

1

2

3

4

5

[prettyName]([input], [offset], [default])

Table 2. OffsetWindowFunctions (Direct Implementations)
OffsetWindowFunction	Description
Lag
Lead

`frame` Lazy Property



frame: WindowFrame

1

2

3

4

5

frame: WindowFrame

Note	`frame` is part of the WindowFunction Contract to define the `WindowFrame` for function expression execution.

frame…FIXME

Verifying Input Data Types — `checkInputDataTypes` Method



checkInputDataTypes(): TypeCheckResult

1

2

3

4

5

checkInputDataTypes(): TypeCheckResult

Note	`checkInputDataTypes` is part of the Expression Contract to verify (check the correctness of) the input data types.

checkInputDataTypes…FIXME

Nondeterministic Contract

2012-09-08admin阅读(1552)

Nondeterministic Expression Contract

Nondeterministic is a contract for Catalyst expressions that are non-deterministic and non-foldable.

Nondeterministic expressions require explicit initialization (with the current partition index) before evaluating a value.



package org.apache.spark.sql.catalyst.expressions

trait Nondeterministic extends Expression {
  // only required methods that have no implementation
  protected def initializeInternal(partitionIndex: Int): Unit
  protected def evalInternal(input: InternalRow): Any
}

1

2

3

4

5

6

7

8

9

10

11

package org.apache.spark.sql.catalyst.expressions

trait Nondeterministic extends Expression {

// only required methods that have no implementation

protected def initializeInternal(partitionIndex: Int): Unit

protected def evalInternal(input: InternalRow): Any

}

Table 1. Nondeterministic Contract
Method	Description
`initializeInternal`	Initializing the `Nondeterministic` expression Used exclusively when `Nondeterministic` expression is requested to initialize
`evalInternal`	Evaluating the `Nondeterministic` expression Used exclusively when `Nondeterministic` expression is requested to evaluate a value

Note	`Nondeterministic` expressions are the target of `PullOutNondeterministic` logical plan rule.

Table 2. Nondeterministic Expressions
Expression	Description
`CurrentBatchTimestamp`
`InputFileBlockLength`
`InputFileBlockStart`
`InputFileName`
MonotonicallyIncreasingID
`NondeterministicExpression`
`Rand`
`Randn`
`RDG`
`SparkPartitionID`

Table 3. Nondeterministic’s Internal Properties (e.g. Registries, Counters and Flags)
Name	Description
deterministic	Always turned off (i.e. `false`)
foldable	Always turned off (i.e. `false`)
`initialized`	Controls whether a `Nondeterministic` expression has been initialized before evaluation. Turned off by default.

Initializing Expression — `initialize` Method



initialize(partitionIndex: Int): Unit

1

2

3

4

5

initialize(partitionIndex: Int): Unit

Internally, initialize initializes itself (with the input partition index) and turns the internal initialized flag on.

Note	`initialize` is used exclusively when InterpretedProjection and `InterpretedMutableProjection` are requested to `initialize` themselves.

Evaluating Expression — `eval` Method



eval(input: InternalRow): Any

1

2

3

4

5

eval(input: InternalRow): Any

Note	`eval` is part of Expression Contract for the interpreted (non-code-generated) expression evaluation, i.e. evaluating a Catalyst expression to a JVM object for a given internal binary row.

eval is just a wrapper of evalInternal that makes sure that initialize has already been executed (and so the expression is initialized).

Internally, eval makes sure that the expression was initialized and calls evalInternal.

eval reports a IllegalArgumentException exception when the internal initialized flag is off, i.e. initialize has not yet been executed.



requirement failed: Nondeterministic expression [name] should be initialized before eval.

1

2

3

4

5

requirement failed: Nondeterministic expression [name] should be initialized before eval.

NamedExpression Contract

2012-09-07admin阅读(1559)

NamedExpression Contract — Catalyst Expressions with Name, ID and Qualifier

NamedExpression is a contract of Catalyst expressions that have a name, exprId, and optional qualifier.



package org.apache.spark.sql.catalyst.expressions

trait NamedExpression extends Expression {
  // only required methods that have no implementation
  def exprId: ExprId
  def name: String
  def newInstance(): NamedExpression
  def qualifier: Option[String]
  def toAttribute: Attribute
}

1

2

3

4

5

6

7

8

9

10

11

12

13

14

package org.apache.spark.sql.catalyst.expressions

trait NamedExpression extends Expression {

// only required methods that have no implementation

def exprId: ExprId

def name: String

def newInstance(): NamedExpression

def qualifier: Option[String]

def toAttribute: Attribute

}

Table 1. NamedExpression Contract
Method	Description
`exprId`	Used when…FIXME
`name`	Used when…FIXME
`qualifier`	Used when…FIXME
`toAttribute`

Creating ExprId — `newExprId` Object Method



newExprId: ExprId

1

2

3

4

5

newExprId: ExprId

newExprId…FIXME

Note	`newExprId` is used when…FIXME

Murmur3Hash

2012-09-06admin阅读(1191)

Murmur3Hash

Murmur3Hash is…FIXME

spark-sql 第33页

ResolvedStar

ResolvedStar

RankLike Contract

RankLike Contract

PrettyAttribute

PrettyAttribute

PlanExpression

PlanExpression Contract for Expressions with Query Plans

ParseToTimestamp

ParseToTimestamp Expression

Creating ParseToTimestamp Instance

ParseToDate

ParseToDate Expression

Creating ParseToDate Instance

OffsetWindowFunction Contract — Unevaluable Window Function Expressions

OffsetWindowFunction Contract — Unevaluable Window Function Expressions

`frame` Lazy Property

Verifying Input Data Types — `checkInputDataTypes` Method

Nondeterministic Contract

Nondeterministic Expression Contract

Initializing Expression — `initialize` Method

Evaluating Expression — `eval` Method

NamedExpression Contract

NamedExpression Contract — Catalyst Expressions with Name, ID and Qualifier

Creating ExprId — `newExprId` Object Method

Murmur3Hash

Murmur3Hash

欢迎关注：spark技术分享

关注公众号：spark技术分享

QQ咨询

回顶部

spark-sql 第33页

ResolvedStar

RankLike Contract

PrettyAttribute

PlanExpression Contract for Expressions with Query Plans

ParseToTimestamp Expression

Creating ParseToTimestamp Instance

ParseToDate Expression

Creating ParseToDate Instance

OffsetWindowFunction Contract — Unevaluable Window Function Expressions

frame Lazy Property

Verifying Input Data Types — checkInputDataTypes Method

Nondeterministic Expression Contract

Initializing Expression — initialize Method

Evaluating Expression — eval Method

NamedExpression Contract — Catalyst Expressions with Name, ID and Qualifier

Creating ExprId — newExprId Object Method

Murmur3Hash

欢迎关注：spark技术分享

关注公众号：spark技术分享

QQ咨询

回顶部

`frame` Lazy Property

Verifying Input Data Types — `checkInputDataTypes` Method

Initializing Expression — `initialize` Method

Evaluating Expression — `eval` Method

Creating ExprId — `newExprId` Object Method